From 8196a806a6f85c1f5a1dc8a59611af9c4b678629 Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 20 Jul 2026 19:52:17 -0700 Subject: [PATCH 1/3] chore(acp): strip stale finding-number references from comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Finding #N numbers cross-referenced a ledger that lives only in the description of an old PR — there is no in-repo registry, so the numbers are meaningless provenance noise. All explanatory prose is preserved; only the "Finding #N:" / "(finding #N)" prefixes and parentheticals are removed. --- crates/buzz-acp/src/acp.rs | 2 +- crates/buzz-acp/src/config.rs | 21 ++++++------ crates/buzz-acp/src/filter.rs | 20 +++++------ crates/buzz-acp/src/lib.rs | 27 +++++++-------- crates/buzz-acp/src/relay.rs | 64 ++++++++++++++++------------------- 5 files changed, 63 insertions(+), 71 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 77fe0f6d73..b553adaabb 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -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"#, ) diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index befb7aa6aa..6f11df1536 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -798,7 +798,7 @@ impl Config { let agent_command = args.agent_command; - // Finding #49a — agent_command must not be empty. + // Validate: 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 +807,7 @@ impl Config { let agent_args = normalize_agent_args(&agent_command, args.agent_args); - // Finding #49b — warn on invalid UUIDs in --channels. + // Warn on invalid UUIDs in --channels. if let Some(ref channels) = args.channels { for ch in channels { if ch.parse::().is_err() { @@ -819,7 +819,7 @@ impl Config { } } - // Finding #49c — cap heartbeat interval at 86400s (24h). + // Cap heartbeat interval at 86400s (24h). let heartbeat_interval = if args.heartbeat_interval > 86400 { tracing::warn!( interval = args.heartbeat_interval, @@ -885,9 +885,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 +1068,7 @@ pub fn load_rules(path: &std::path::Path) -> Result, Confi ))); } - // Finding #49d — warn when Config mode has no rules; agent will receive nothing. + // Warn when Config mode has no rules; agent will receive nothing. if config.rules.is_empty() { tracing::warn!( path = %path.display(), @@ -1099,7 +1099,7 @@ pub fn load_rules(path: &std::path::Path) -> Result, 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. + // 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 +1121,8 @@ pub fn load_rules(path: &std::path::Path) -> Result, Confi ))); } } - // Initialise the consecutive-timeout counter (finding #25). - // Deserialization leaves it at default (new Arc) - // but we set it explicitly here for clarity. + // Initialise the consecutive-timeout counter. Deserialization leaves it at + // default (new Arc) but we set it explicitly here for clarity. rule.consecutive_timeouts = Arc::new(AtomicU32::new(0)); } diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index b256e36f10..43edd969dd 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -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, - /// 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>, - /// 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> = 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> = /// - 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(); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 28bb774008..5f0d5e499c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -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> = 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 => { diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 67d01fd38d..90111f494e 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -474,7 +474,7 @@ enum RelayCommand { SubscribeObserverControls, /// Publish a signed event to the relay (for typing indicators, etc.). PublishEvent { event: Box }, - /// Set the startup watermark timestamp for Finding #22. + /// Set the startup watermark timestamp. /// The background task uses this as the floor `since` for membership /// notification replay so events before startup are never re-delivered. SetStartupWatermark { ts: u64 }, @@ -544,8 +544,8 @@ 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. + // 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,7 +811,7 @@ impl HarnessRelay { Ok(event) } - /// Set the startup watermark timestamp (Finding #22). + /// Set the startup watermark timestamp. /// /// Call this once after `connect()` with the Unix timestamp captured just /// before the relay connection was established. The background task uses @@ -945,9 +945,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, /// Replay floor captured when each channel was first subscribed. /// Used as the `since` fallback on reconnect for channels that have no @@ -1235,7 +1235,7 @@ async fn run_background_task( ) { let mut state = BgState::new(); - // Finding #8: process any messages buffered during the initial auth handshake. + // 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, @@ -1301,18 +1301,18 @@ 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! + // 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 +1380,7 @@ 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. + // Track pong replies directly, before dispatch. if matches!(msg, Message::Pong(_)) { last_pong = Instant::now(); ping_sent = false; @@ -1604,8 +1604,7 @@ 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. + // Log once when the connection has been stable for STABLE_CONNECTION_SECS. Diagnostic only. if !stable_logged && connected_since.elapsed() > Duration::from_secs(STABLE_CONNECTION_SECS) { stable_logged = true; @@ -1682,7 +1681,7 @@ async fn handle_ws_message( channel_id: channel_uuid, 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) { @@ -1706,8 +1705,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 +1723,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 +1746,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. state.proactive_resubscribe_needed = true; warn!( channel_id = %channel_id, @@ -1788,7 +1786,7 @@ async fn handle_ws_message( return true; } - // Finding #15: CLOSED needs cleanup and resubscribe, not just logging. + // CLOSED needs cleanup and resubscribe, not just logging. // Classify the error to decide how to respond. let is_auth_error = message.starts_with("auth-required") || message.starts_with("restricted") @@ -1885,7 +1883,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 +1898,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 +1923,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 @@ -2067,9 +2065,8 @@ async fn resubscribe_after_reconnect( /// 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. +/// 5 attempts with 1s→2s→4s→8s→16s backoff (was 3 attempts), ±20% jitter on each sleep. +/// Processes the handshake buffer on success. /// /// Outcome of an autonomous reconnect attempt. enum ReconnectOutcome { @@ -2150,10 +2147,9 @@ async fn try_autonomous_reconnect( observer_control_tx: &mpsc::Sender, 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 +2162,7 @@ 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. + // Process buffered messages from the handshake. let handshake_ok = process_handshake_buffer( ws, handshake_buffer, @@ -2263,8 +2259,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 +2277,7 @@ 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. + // Process buffered messages from the handshake. let handshake_ok = process_handshake_buffer( ws, handshake_buffer, From 5ea7fa02131d94414380cebcb58073d2c85b8e20 Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 20 Jul 2026 20:14:10 -0700 Subject: [PATCH 2/3] chore(acp): delete narration comments, keep only non-derivable constraints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second pass over the finding-number cleanup: comments that merely narrated the next line (Validate: agent_command must not be empty; Warn at 80% capacity; Process any messages buffered...; Log once when stable...) are deleted outright. Comments that state constraints or rationale the code cannot show — idle_timeout must be strictly less than max_turn_duration or it becomes a dead letter; proactively trigger resubscribe without waiting for a disconnect; recv() None means all senders dropped — are kept. The misplaced function description on ReconnectOutcome's doc comment is also corrected. --- crates/buzz-acp/src/config.rs | 8 +------- crates/buzz-acp/src/relay.rs | 30 ++++++------------------------ 2 files changed, 7 insertions(+), 31 deletions(-) diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 6f11df1536..d78c4fa319 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -798,7 +798,6 @@ impl Config { let agent_command = args.agent_command; - // Validate: 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); - // Warn on invalid UUIDs in --channels. if let Some(ref channels) = args.channels { for ch in channels { if ch.parse::().is_err() { @@ -819,7 +817,6 @@ impl Config { } } - // Cap heartbeat interval at 86400s (24h). let heartbeat_interval = if args.heartbeat_interval > 86400 { tracing::warn!( interval = args.heartbeat_interval, @@ -1068,7 +1065,6 @@ pub fn load_rules(path: &std::path::Path) -> Result, Confi ))); } - // 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, Confi } // Fail fast: parse the expression at load time so typos don't // silently produce dead rules at runtime. - // 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,8 +1116,7 @@ pub fn load_rules(path: &std::path::Path) -> Result, Confi ))); } } - // Initialise the consecutive-timeout counter. Deserialization leaves it at - // default (new Arc) 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)); } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 90111f494e..5035acd83e 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -474,9 +474,7 @@ enum RelayCommand { SubscribeObserverControls, /// Publish a signed event to the relay (for typing indicators, etc.). PublishEvent { event: Box }, - /// Set the startup watermark timestamp. - /// 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`. - // 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. + /// 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 }) @@ -1235,8 +1230,6 @@ async fn run_background_task( ) { let mut state = BgState::new(); - // 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, @@ -1380,7 +1373,6 @@ async fn run_background_task( // Determine if the socket is lost. let socket_lost = match raw { Some(Ok(msg)) => { - // Track pong replies directly, before dispatch. if matches!(msg, Message::Pong(_)) { last_pong = Instant::now(); ping_sent = false; @@ -1604,7 +1596,6 @@ async fn run_background_task( } } - // Log once when the connection has been stable for STABLE_CONNECTION_SECS. Diagnostic only. if !stable_logged && connected_since.elapsed() > Duration::from_secs(STABLE_CONNECTION_SECS) { stable_logged = true; @@ -1681,7 +1672,6 @@ async fn handle_ws_message( channel_id: channel_uuid, event: *event, }; - // Warn at 80% capacity. let cap = event_tx.max_capacity(); let used = cap - event_tx.capacity(); if used >= (cap * 4 / 5) { @@ -1746,7 +1736,7 @@ async fn handle_ws_message( .entry(channel_id) .and_modify(|d| *d = (*d).min(ts)) .or_insert(ts); - // Proactively trigger resubscribe. + // Proactively trigger resubscribe without waiting for a disconnect. state.proactive_resubscribe_needed = true; warn!( channel_id = %channel_id, @@ -1787,7 +1777,6 @@ async fn handle_ws_message( } // CLOSED needs cleanup and resubscribe, not just logging. - // Classify the error to decide how to respond. let is_auth_error = message.starts_with("auth-required") || message.starts_with("restricted") || message.contains("auth"); @@ -2063,11 +2052,6 @@ async fn resubscribe_after_reconnect( all_ok } -/// Attempt autonomous reconnect on socket loss. -/// -/// 5 attempts with 1s→2s→4s→8s→16s backoff (was 3 attempts), ±20% jitter on each sleep. -/// Processes the handshake buffer on success. -/// /// Outcome of an autonomous reconnect attempt. enum ReconnectOutcome { /// Reconnected and resubscribed successfully. @@ -2162,7 +2146,6 @@ async fn try_autonomous_reconnect( Ok((new_ws, handshake_buffer)) => { *ws = new_ws; info!("autonomous reconnect succeeded (attempt {})", attempt + 1); - // Process buffered messages from the handshake. let handshake_ok = process_handshake_buffer( ws, handshake_buffer, @@ -2277,7 +2260,6 @@ async fn wait_for_reconnect( Ok((new_ws, handshake_buffer)) => { *ws = new_ws; info!("relay reconnected to {relay_url}"); - // Process buffered messages from the handshake. let handshake_ok = process_handshake_buffer( ws, handshake_buffer, From 75041dadd7180cf64714c641c16ea7528bfc9eae Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 20 Jul 2026 20:24:43 -0700 Subject: [PATCH 3/3] chore(acp): drop final positional comment in relay background loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit "Check proactive resubscribe flag before blocking on select!" narrates the position of the check rather than its rationale — the loop structure makes the ordering self-evident. --- crates/buzz-acp/src/relay.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 5035acd83e..dee8359a77 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -1305,7 +1305,6 @@ async fn run_background_task( let mut stable_logged = false; loop { - // 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");