External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore(acp): strip stale finding-number references from comments (#2202)

Browse Source
This commit is contained in:
Will Pfleger
2026-07-20 23:28:47 -04:00
committed by GitHub
parent 0638009b32
commit 0791fff8d2
5 changed files with 49 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-acp/src/acp.rs
+1 -1
View File
@@ -2655,7 +2655,7 @@ mod tests {
#[tokio::test]
async fn idle_resets_on_stdout_activity() {
// Send valid JSON (session/update notifications) to reset the idle timer.
// Non-JSON lines no longer reset idle (Finding #6 hardening).
// Non-JSON lines no longer reset idle — only valid JSON notifications do.
let mut client = spawn_script(
r#"for i in $(seq 1 10); do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"agent_thought_chunk","content":{"text":"thinking"}}}}'; sleep 0.05; done; sleep 10"#,
)
crates/buzz-acp/src/config.rs
+4 -11
View File
@@ -798,7 +798,6 @@ impl Config {
let agent_command = args.agent_command;
// Finding #49a — agent_command must not be empty.
if agent_command.trim().is_empty() {
return Err(ConfigError::ConfigFile(
"agent_command must not be empty".into(),
@@ -807,7 +806,6 @@ impl Config {
let agent_args = normalize_agent_args(&agent_command, args.agent_args);
// Finding #49b — warn on invalid UUIDs in --channels.
if let Some(ref channels) = args.channels {
for ch in channels {
if ch.parse::<Uuid>().is_err() {
@@ -819,7 +817,6 @@ impl Config {
}
}
// Finding #49c — cap heartbeat interval at 86400s (24h).
let heartbeat_interval = if args.heartbeat_interval > 86400 {
tracing::warn!(
interval = args.heartbeat_interval,
@@ -885,9 +882,9 @@ impl Config {
}
};
// Finding #20 — idle_timeout must be strictly less than max_turn_duration.
// If idle_timeout >= max_turn_duration, the absolute wall-clock cap would
// fire before the idle timeout ever could, making idle_timeout a dead letter.
// idle_timeout must be strictly less than max_turn_duration. If idle_timeout
// >= max_turn_duration, the absolute wall-clock cap would fire before the idle
// timeout ever could, making idle_timeout a dead letter.
if idle_timeout_secs >= max_turn_duration_secs {
return Err(ConfigError::ConfigFile(format!(
"idle_timeout ({}s) must be less than max_turn_duration ({}s)",
@@ -1068,7 +1065,6 @@ pub fn load_rules(path: &std::path::Path) -> Result<Vec<SubscriptionRule>, Confi
)));
}
// Finding #49d — warn when Config mode has no rules; agent will receive nothing.
if config.rules.is_empty() {
tracing::warn!(
path = %path.display(),
@@ -1099,7 +1095,6 @@ pub fn load_rules(path: &std::path::Path) -> Result<Vec<SubscriptionRule>, Confi
}
// Fail fast: parse the expression at load time so typos don't
// silently produce dead rules at runtime.
// Finding #34 — store the compiled AST so match_event never re-parses.
match evalexpr::build_operator_tree(expr) {
Ok(node) => {
rule.compiled_filter = Some(Arc::new(node));
@@ -1121,9 +1116,7 @@ pub fn load_rules(path: &std::path::Path) -> Result<Vec<SubscriptionRule>, Confi
)));
}
}
// Initialise the consecutive-timeout counter (finding #25).
// Deserialization leaves it at default (new Arc<AtomicU32::new(0)>)
// but we set it explicitly here for clarity.
// Deserialization leaves consecutive_timeouts at its zero default; reset explicitly.
rule.consecutive_timeouts = Arc::new(AtomicU32::new(0));
}
crates/buzz-acp/src/filter.rs
+10 -10
View File
@@ -97,14 +97,14 @@ pub struct SubscriptionRule {
/// Tag passed to the prompt template. Falls back to `name` if absent.
#[serde(default)]
pub prompt_tag: Option<String>,
/// Pre-compiled evalexpr AST for the `filter` expression (finding #34).
/// Pre-compiled evalexpr AST for the `filter` expression.
///
/// Populated by `load_rules()` at startup so `match_event` never re-parses
/// the expression string on the hot path. `None` when `filter` is `None`
/// or the rule was constructed without calling `load_rules()` (e.g. tests).
#[serde(skip)]
pub compiled_filter: Option<Arc<evalexpr::Node>>,
/// Consecutive filter-evaluation timeout counter (finding #25).
/// Consecutive filter-evaluation timeout counter.
///
/// Incremented on each timeout; reset on any successful evaluation.
/// When this reaches `MAX_CONSECUTIVE_TIMEOUTS`, the rule is treated as
@@ -164,7 +164,7 @@ const MAX_EXPR_LEN: usize = 4096;
/// Maximum wall-clock time allowed for a single evalexpr evaluation.
const EVAL_TIMEOUT: Duration = Duration::from_millis(100);
/// Maximum concurrent blocking filter evaluations (finding #13 / Issue 3).
/// Maximum concurrent blocking filter evaluations.
///
/// The semaphore permit is moved into each `spawn_blocking` closure so it is
/// held until the blocking thread finishes — not just until the caller's timeout
@@ -178,7 +178,7 @@ const MAX_CONCURRENT_FILTER_EVALS: usize = 4;
/// `OwnedSemaphorePermit` that can be moved into the `spawn_blocking` closure.
/// This ensures the permit is held until the blocking task actually finishes —
/// not just until the caller's timeout fires — so the semaphore truly bounds
/// the number of live blocking threads (finding #13 / Issue 3).
/// the number of live blocking threads.
static FILTER_EVAL_SEMAPHORE: std::sync::LazyLock<Arc<tokio::sync::Semaphore>> =
std::sync::LazyLock::new(|| Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_FILTER_EVALS)));
@@ -187,11 +187,11 @@ static FILTER_EVAL_SEMAPHORE: std::sync::LazyLock<Arc<tokio::sync::Semaphore>> =
/// - Caps expression length at [`MAX_EXPR_LEN`] bytes.
/// - Acquires an owned permit from [`FILTER_EVAL_SEMAPHORE`] and moves it into
/// the blocking closure so it is held until the task finishes, not just until
/// the caller's timeout fires (finding #13 / Issue 3).
/// the caller's timeout fires.
/// - Runs evaluation on a blocking thread with a [`EVAL_TIMEOUT`] hard timeout.
/// - When a pre-compiled `node` is provided (via `Arc`), uses
/// `node.eval_boolean_with_context()` instead of re-parsing the expression
/// string on every call (finding #34).
/// string on every call.
/// - Registers custom string helpers: `str_contains`, `str_starts_with`,
/// `str_ends_with`, `str_len` (duplicated intentionally from buzz-workflow).
pub async fn evaluate_filter(
@@ -212,7 +212,7 @@ pub async fn evaluate_filter(
// Acquire an *owned* permit so it can be moved into the spawn_blocking closure.
// The permit is held until the blocking task actually completes — not just until
// the caller's timeout fires — so the semaphore truly bounds the number of live
// blocking threads even when callers time out (Issue 3 / finding #13).
// blocking threads even when callers time out.
//
// The acquire itself is bounded by EVAL_TIMEOUT: if all permits are held by
// wedged blocking tasks, we time out instead of blocking the main event loop.
@@ -351,10 +351,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 (finding #45).
/// access.
/// 4. **filter** — if `Some`, the evalexpr expression must evaluate to `true`.
///
/// # Fail-closed filter error handling (finding #25)
/// # Fail-closed filter error handling
///
/// Any filter evaluation error — including timeout — causes the **entire
/// `match_event` call** to return `None` (no match for any rule). We never
@@ -386,7 +386,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 (finding #45) — avoids relying on the Display impl of tag kind.
// access — avoids relying on the Display impl of tag kind.
if rule.require_mention {
let mentioned = event.tags.iter().any(|tag| {
let s = tag.as_slice();
crates/buzz-acp/src/lib.rs
+12 -15
View File
@@ -1157,11 +1157,10 @@ async fn tokio_main() -> Result<()> {
);
}
//
// Finding #10: one agent failing to start must not kill the whole pool.
// We attempt each spawn under a 60-second timeout; failures are logged and
// skipped. If ALL agents fail we return an error. A partial pool is valid —
// the harness continues with reduced capacity and logs a warning.
// One agent failing to start must not kill the whole pool. We attempt each
// spawn under a 60-second timeout; failures are logged and skipped. If ALL
// agents fail we return an error. A partial pool is valid — the harness
// continues with reduced capacity and logs a warning.
let mut agent_slots: Vec<Option<OwnedAgent>> = Vec::with_capacity(config.agents as usize);
for i in 0..config.agents as usize {
// Spawn OUTSIDE the timeout so we always own the child for cleanup.
@@ -1248,13 +1247,11 @@ async fn tokio_main() -> Result<()> {
tracing::info!("agent_pool_ready agents={}", live_count);
let mut pool = AgentPool::from_slots(agent_slots);
//
// Finding #22: capture a startup watermark BEFORE connecting to the relay.
// This timestamp is used for membership notification replay (via
// startup_watermark) and as the initial subscribe_since for channels
// discovered at startup. The Subscribe handler falls back to
// subscribe_since when last_seen is None, closing the blind spot
// between "agents ready" and "first REQ sent".
// Capture a startup watermark BEFORE connecting to the relay. This timestamp
// is used for membership notification replay (via startup_watermark) and as
// the initial subscribe_since for channels discovered at startup. The Subscribe
// handler falls back to subscribe_since when last_seen is None, closing the
// blind spot between "agents ready" and "first REQ sent".
let startup_watermark: u64 = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
@@ -1273,7 +1270,7 @@ async fn tokio_main() -> Result<()> {
.await
.map_err(|e| anyhow::anyhow!("relay connect error: {e}"))?;
// Finding #22: tell the relay background task the watermark so it can use
// Tell the relay background task the watermark so it can use
// `since = watermark - 5s` on the first REQ instead of `since=now`.
// Best-effort: a failure here is non-fatal (we just lose the startup window
// protection, which is the same as the pre-fix behaviour).
@@ -1697,8 +1694,8 @@ async fn tokio_main() -> Result<()> {
let (result_rx, join_set) = pool.rx_and_join_set();
tokio::select! {
biased;
// Finding #24: recv() returning None means all senders dropped
// (pool was torn down). Break cleanly instead of panicking.
// recv() returning None means all senders dropped (pool was torn down).
// Break cleanly instead of panicking.
r = result_rx.recv() => match r {
Some(result) => Some(PoolEvent::Result(Box::new(result))),
None => {
crates/buzz-acp/src/relay.rs
+22 -45
View File
@@ -474,9 +474,7 @@ enum RelayCommand {
SubscribeObserverControls,
/// Publish a signed event to the relay (for typing indicators, etc.).
PublishEvent { event: Box<Event> },
/// Set the startup watermark timestamp for Finding #22.
/// The background task uses this as the floor `since` for membership
/// notification replay so events before startup are never re-delivered.
/// Floor `since` for membership notification replay; events before startup are never re-delivered.
SetStartupWatermark { ts: u64 },
}
@@ -544,8 +542,6 @@ impl HarnessRelay {
// jittered backoff. A terminal error (bad URL, bad auth tag,
// rejected/invalid signing key) fails immediately — see
// `is_terminal_connect_error`.
// Finding #8: capture the handshake buffer and pass it to the background
// task so buffered messages aren't silently discarded.
let (ws, handshake_buffer) =
retry_initial_connect(|| do_connect(relay_url, keys, auth_tag.as_ref())).await?;
@@ -811,12 +807,11 @@ impl HarnessRelay {
Ok(event)
}
/// Set the startup watermark timestamp (Finding #22).
/// Pins the floor `since` for membership notification replay.
///
/// Call this once after `connect()` with the Unix timestamp captured just
/// before the relay connection was established. The background task uses
/// this as the floor `since` for membership notification replay so events
/// predating this session are never re-delivered after reconnect.
/// Call once after `connect()` with the Unix timestamp captured just before
/// the relay connection was established. The background task uses this so
/// events predating this session are never re-delivered after reconnect.
pub async fn set_startup_watermark(&self, ts: u64) -> Result<(), RelayError> {
self.cmd_tx
.send(RelayCommand::SetStartupWatermark { ts })
@@ -945,9 +940,9 @@ struct BgState {
/// The main loop checks this flag and triggers a proactive resubscribe
/// (without waiting for a disconnect) so dropped events are replayed.
proactive_resubscribe_needed: bool,
/// Unix timestamp captured just before the relay connection was established
/// (Finding #22). Used as the floor `since` for membership notification
/// replay so events predating this session are never re-delivered.
/// Unix timestamp captured just before the relay connection was established.
/// Used as the floor `since` for membership notification replay so events
/// predating this session are never re-delivered.
startup_watermark: Option<u64>,
/// Replay floor captured when each channel was first subscribed.
/// Used as the `since` fallback on reconnect for channels that have no
@@ -1235,8 +1230,6 @@ async fn run_background_task(
) {
let mut state = BgState::new();
// Finding #8: process any messages buffered during the initial auth handshake.
// If a buffered message signals connection drop, trigger reconnect immediately.
let handshake_ok = process_handshake_buffer(
&mut ws,
initial_handshake_buffer,
@@ -1301,18 +1294,17 @@ async fn run_background_task(
// no reset needed here since they haven't been declared yet.
}
// Finding #31: client-initiated ping to detect silent connection death.
// Client-initiated ping to detect silent connection death.
let mut ping_interval = tokio::time::interval(PING_INTERVAL);
ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut last_pong = Instant::now();
let mut ping_sent = false;
// Finding #42: track connection stability for backoff reset.
// Track connection stability for backoff reset.
let mut connected_since = Instant::now();
let mut stable_logged = false;
loop {
// Finding #3: check proactive resubscribe flag before blocking on select!
if state.proactive_resubscribe_needed {
state.proactive_resubscribe_needed = false;
info!("proactive resubscribe triggered by backpressure event loss");
@@ -1380,7 +1372,6 @@ async fn run_background_task(
// Determine if the socket is lost.
let socket_lost = match raw {
Some(Ok(msg)) => {
// Finding #31: track pong replies directly, before dispatch.
if matches!(msg, Message::Pong(_)) {
last_pong = Instant::now();
ping_sent = false;
@@ -1604,8 +1595,6 @@ async fn run_background_task(
}
}
// Finding #42: log when connection has been stable for STABLE_CONNECTION_SECS.
// Log once when the connection has been stable. Diagnostic only.
if !stable_logged && connected_since.elapsed() > Duration::from_secs(STABLE_CONNECTION_SECS)
{
stable_logged = true;
@@ -1682,7 +1671,6 @@ async fn handle_ws_message(
channel_id: channel_uuid,
event: *event,
};
// Finding #3: warn at 80% capacity.
let cap = event_tx.max_capacity();
let used = cap - event_tx.capacity();
if used >= (cap * 4 / 5) {
@@ -1706,8 +1694,7 @@ async fn handle_ws_message(
// replay starts early enough to re-deliver it.
state.membership_dropped_since =
Some(state.membership_dropped_since.map_or(ts, |d| d.min(ts)));
// Finding #3: proactively trigger resubscribe without
// waiting for a disconnect.
// Proactively trigger resubscribe without waiting for a disconnect.
state.proactive_resubscribe_needed = true;
warn!(
channel_id = %channel_uuid,
@@ -1725,7 +1712,7 @@ async fn handle_ws_message(
channel_id,
event: *event,
};
// Finding #3: warn at 80% capacity.
// Warn at 80% capacity.
let cap = event_tx.max_capacity();
let used = cap - event_tx.capacity();
if used >= (cap * 4 / 5) {
@@ -1748,7 +1735,7 @@ async fn handle_ws_message(
.entry(channel_id)
.and_modify(|d| *d = (*d).min(ts))
.or_insert(ts);
// Finding #3: proactively trigger resubscribe.
// Proactively trigger resubscribe without waiting for a disconnect.
state.proactive_resubscribe_needed = true;
warn!(
channel_id = %channel_id,
@@ -1788,8 +1775,7 @@ async fn handle_ws_message(
return true;
}
// Finding #15: CLOSED needs cleanup and resubscribe, not just logging.
// Classify the error to decide how to respond.
// CLOSED needs cleanup and resubscribe, not just logging.
let is_auth_error = message.starts_with("auth-required")
|| message.starts_with("restricted")
|| message.contains("auth");
@@ -1885,7 +1871,7 @@ async fn handle_ws_message(
}
}
RelayMessage::Auth { challenge } => {
// Finding #18: AUTH send failure must trigger reconnect.
// AUTH send failure must trigger reconnect.
debug!("received mid-session AUTH challenge — re-authenticating");
if let Err(e) =
send_auth_response(ws, &challenge, relay_url, keys, auth_tag).await
@@ -1900,7 +1886,7 @@ async fn handle_ws_message(
message,
} => {
if !accepted && message.starts_with("auth") {
// Finding #18: AUTH OK with accepted=false means auth was rejected.
// AUTH OK with accepted=false means auth was rejected.
warn!("mid-session AUTH rejected (event {event_id}): {message} — triggering reconnect");
return false;
}
@@ -1925,7 +1911,7 @@ async fn handle_ws_message(
}
}
/// Process messages buffered during the NIP-42 auth handshake (Finding #8).
/// Process messages buffered during the NIP-42 auth handshake.
///
/// `do_connect` buffers any non-AUTH/non-OK messages it receives while waiting
/// for the challenge and OK. Those messages would otherwise be silently
@@ -2065,12 +2051,6 @@ async fn resubscribe_after_reconnect(
all_ok
}
/// Attempt autonomous reconnect on socket loss.
///
/// Finding #42: 5 attempts with 1s→2s→4s→8s→16s backoff (was 3 attempts).
/// Finding #27: ±20% jitter on each sleep.
/// Finding #8: process handshake buffer on success.
///
/// Outcome of an autonomous reconnect attempt.
enum ReconnectOutcome {
/// Reconnected and resubscribed successfully.
@@ -2150,10 +2130,9 @@ async fn try_autonomous_reconnect(
observer_control_tx: &mpsc::Sender<Event>,
auth_tag: Option<&nostr::Tag>,
) -> ReconnectOutcome {
// Finding #42: 5 attempts, up to 16s base backoff. Shares delay values
// with the initial-connect retry in `HarnessRelay::connect()`
// (STARTUP_CONNECT_BACKOFFS) — see its doc comment for how the two
// loops consume the array differently.
// 5 attempts, up to 16s base backoff. Shares delay values with the
// initial-connect retry in `HarnessRelay::connect()` (STARTUP_CONNECT_BACKOFFS) —
// see its doc comment for how the two loops consume the array differently.
let backoffs = STARTUP_CONNECT_BACKOFFS;
for (attempt, delay) in backoffs.iter().enumerate() {
@@ -2166,7 +2145,6 @@ async fn try_autonomous_reconnect(
Ok((new_ws, handshake_buffer)) => {
*ws = new_ws;
info!("autonomous reconnect succeeded (attempt {})", attempt + 1);
// Finding #8: process buffered messages from the handshake.
let handshake_ok = process_handshake_buffer(
ws,
handshake_buffer,
@@ -2263,8 +2241,8 @@ async fn wait_for_reconnect(
}
}
// Finding #42: 6 attempts with backoff up to 32s + jitter (Finding #27).
// Finding #27: use tokio::select! so shutdown is honoured during sleep.
// 6 attempts with backoff up to 32s + jitter; uses tokio::select! so shutdown is
// honoured during sleep.
let backoffs = [
Duration::from_secs(1),
Duration::from_secs(2),
@@ -2281,7 +2259,6 @@ async fn wait_for_reconnect(
Ok((new_ws, handshake_buffer)) => {
*ws = new_ws;
info!("relay reconnected to {relay_url}");
// Finding #8: process buffered messages from the handshake.
let handshake_ok = process_handshake_buffer(
ws,
handshake_buffer,