diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 403512a322..627aff79b9 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1964,6 +1964,15 @@ async fn tokio_main() -> Result<()> { // stripped for a legitimately re-added channel. removed_channels.remove(&ch); + // Converge with the relay layer before the + // "already subscribed" guard runs: a channel + // the relay dropped is not subscribed, no + // matter what our intent-only set says. + clear_dropped_channels( + &mut subscribed_channel_ids, + relay.take_dropped_channels(), + ); + if subscribed_channel_ids.contains(&ch) { tracing::debug!(channel_id = %ch, "membership notification: channel already subscribed"); } else if let Some(filter) = config::resolve_dynamic_channel_filter(&config, ch, &rules) { @@ -3606,6 +3615,112 @@ fn dispatch_heartbeat( tracing::info!(agent = agent_index, "heartbeat_fired"); } +/// Drop relay-dropped channels from the harness's subscription view. +/// +/// `subscribed_channel_ids` records the *intent* to subscribe, not a confirmed +/// subscription: `subscribe_channel*` only queues a REQ, and the relay layer can +/// tear a channel down afterwards (access denied, missing filter state) without +/// the main loop hearing about it. Left uncleared, the "already subscribed" +/// guard on the member-added path turns every later membership notification for +/// that channel into a no-op and the channel stays dead until restart. +/// +/// Returns the channels that were actually cleared. +fn clear_dropped_channels( + subscribed_channel_ids: &mut HashSet, + dropped: impl IntoIterator, +) -> Vec { + let mut cleared = Vec::new(); + for ch in dropped { + if subscribed_channel_ids.remove(&ch) { + tracing::warn!( + channel_id = %ch, + "relay dropped this channel's subscription — cleared local subscribe state so a membership notification can re-subscribe" + ); + cleared.push(ch); + } + } + cleared +} + +#[cfg(test)] +mod dropped_channel_resubscribe_tests { + use super::clear_dropped_channels; + use std::collections::HashSet; + use uuid::Uuid; + + /// The member-added guard from the main loop: a channel already in the set + /// is treated as live and no subscribe is issued. + fn would_subscribe(subscribed: &HashSet, ch: Uuid) -> bool { + !subscribed.contains(&ch) + } + + #[test] + fn dropped_channel_is_resubscribed_on_next_membership_notification() { + let ch = Uuid::new_v4(); + let mut subscribed = HashSet::from([ch]); + + // Before the relay drops it, a membership notification is a correct no-op. + assert!( + !would_subscribe(&subscribed, ch), + "a live channel must not be resubscribed on every membership event" + ); + + // Relay layer drops the channel (access denied / missing filter) and + // reports it back through take_dropped_channels(). + let cleared = clear_dropped_channels(&mut subscribed, [ch]); + + assert_eq!( + cleared, + vec![ch], + "the drop must clear local subscribe state" + ); + assert!( + would_subscribe(&subscribed, ch), + "after a drop, the next membership notification must re-subscribe, not no-op" + ); + } + + #[test] + fn channels_that_were_not_dropped_are_untouched() { + let dropped_ch = Uuid::new_v4(); + let live_ch = Uuid::new_v4(); + let mut subscribed = HashSet::from([dropped_ch, live_ch]); + + clear_dropped_channels(&mut subscribed, [dropped_ch]); + + assert!(!subscribed.contains(&dropped_ch)); + assert!( + subscribed.contains(&live_ch), + "an unrelated channel must keep its subscription state" + ); + } + + #[test] + fn drop_report_for_an_unknown_channel_is_a_no_op() { + let mut subscribed = HashSet::new(); + let cleared = clear_dropped_channels(&mut subscribed, [Uuid::new_v4()]); + assert!( + cleared.is_empty(), + "a drop for a channel we never tracked must clear nothing" + ); + } + + #[test] + fn resubscribe_then_second_drop_clears_again() { + let ch = Uuid::new_v4(); + let mut subscribed = HashSet::from([ch]); + + clear_dropped_channels(&mut subscribed, [ch]); + // Membership notification re-subscribes. + subscribed.insert(ch); + assert!(!would_subscribe(&subscribed, ch)); + + // A second drop must clear it again — the fix is not one-shot. + clear_dropped_channels(&mut subscribed, [ch]); + assert!(would_subscribe(&subscribed, ch)); + } +} + #[cfg(test)] mod agent_draft_prompt_tests { #[test] diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index aea5cee077..d653796f48 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -22,8 +22,19 @@ //! channel. `next_event()` reads from the event receiver. use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{Arc, Mutex}; use std::time::Duration; +/// Channels the relay layer has permanently dropped, shared between the +/// background task (writer) and [`HarnessRelay`] (draining reader). +/// +/// The background task tears a channel subscription down in several places and +/// never restores it (access denied by the relay, missing filter state). The +/// harness main loop keeps its own `subscribed_channel_ids` view; without this +/// hand-off that view never converges and every later membership notification +/// for the channel is a silent no-op until the process restarts. +pub(crate) type DroppedChannels = Arc>>; + /// Default capacity of the event channel from background task to harness. /// Override with `BUZZ_ACP_EVENT_BUFFER` env var at startup. const EVENT_CHANNEL_CAPACITY_DEFAULT: usize = 256; @@ -551,6 +562,8 @@ pub struct HarnessRelay { observer_control_rx: Option>, /// Sender for commands to the background task. cmd_tx: mpsc::Sender, + /// Channels the background task has dropped, drained by the harness. + dropped_channels: DroppedChannels, /// HTTP client for HTTP bridge calls. http: reqwest::Client, /// WebSocket URL of the relay. @@ -625,6 +638,8 @@ impl HarnessRelay { mpsc::channel::(event_channel_capacity()); let (cmd_tx, cmd_rx) = mpsc::channel::(CMD_CHANNEL_CAPACITY); + let dropped_channels: DroppedChannels = DroppedChannels::default(); + let bg_dropped_channels = dropped_channels.clone(); let bg_keys = keys.clone(); let bg_relay_url = relay_url.to_string(); let bg_agent_pubkey_hex = agent_pubkey_hex.to_string(); @@ -641,6 +656,7 @@ impl HarnessRelay { bg_relay_url, bg_agent_pubkey_hex, bg_auth_tag, + bg_dropped_channels, ) .await; }); @@ -649,6 +665,7 @@ impl HarnessRelay { event_rx, observer_control_rx: Some(observer_control_rx), cmd_tx, + dropped_channels, http: reqwest::Client::builder() .timeout(std::time::Duration::from_secs(10)) .connect_timeout(std::time::Duration::from_secs(5)) @@ -817,6 +834,22 @@ impl HarnessRelay { Ok(()) } + /// Drain the channels the relay layer has dropped since the last call. + /// + /// Each dropped channel is returned exactly once. The harness clears these + /// from its own subscription bookkeeping so a later membership notification + /// re-subscribes the channel instead of short-circuiting as "already + /// subscribed". + pub fn take_dropped_channels(&self) -> Vec { + match self.dropped_channels.lock() { + Ok(mut dropped) => dropped.drain().collect(), + // A poisoned lock means a writer panicked mid-insert. Reporting no + // drops is the safe direction: the harness keeps its current view + // rather than spuriously re-subscribing. + Err(_) => Vec::new(), + } + } + /// Wait for the next event from any subscribed channel. /// /// Reads from the background task's event channel. Returns `None` on @@ -1064,6 +1097,9 @@ struct BgState { /// A single failed channel REQ is parked here instead of aborting the whole /// reconnect. Drained by the main loop. Flushed on each reconnect attempt. resubscribe_retry: HashSet, + /// Channels dropped by this task, handed back to the harness main loop so + /// its `subscribed_channel_ids` view converges with relay reality. + dropped_channels: DroppedChannels, /// Current position in the exponential backoff ladder. /// /// Persisted across calls to `wait_for_reconnect` so a flapping link stays at @@ -1075,6 +1111,7 @@ struct BgState { impl BgState { fn new() -> Self { Self { + dropped_channels: DroppedChannels::default(), active_subscriptions: HashMap::new(), last_seen: HashMap::new(), seen_ids: TwoGenDedup::new(SEEN_ID_LIMIT), @@ -1143,6 +1180,34 @@ impl BgState { /// Clear all per-channel state for a channel that is being unsubscribed. /// Prevents stale replay on re-subscribe and avoids unbounded state growth /// for channels that are removed and never re-added. + /// Build a state that reports its channel drops through `dropped_channels`. + fn with_dropped_channels(dropped_channels: DroppedChannels) -> Self { + Self { + dropped_channels, + ..Self::new() + } + } + + /// Record a channel this task has given up on. + /// + /// The relay layer never restores these on its own, so the harness must be + /// told: it clears its local subscribe state and re-subscribes on the next + /// membership notification. Logged at warn so the failure mode is visible + /// in agent logs instead of presenting as a silently dead channel. + fn record_channel_drop(&mut self, channel_id: Uuid, reason: &str) { + warn!( + "channel {channel_id} dropped by relay layer ({reason}) - harness will re-subscribe on the next membership notification" + ); + match self.dropped_channels.lock() { + Ok(mut dropped) => { + dropped.insert(channel_id); + } + Err(_) => { + warn!("dropped-channel set poisoned - channel {channel_id} drop not reported") + } + } + } + fn clear_channel_state(&mut self, channel_id: &Uuid) { self.last_seen.remove(channel_id); self.subscribe_since.remove(channel_id); @@ -1261,6 +1326,12 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { filter, replay_since, } => { + // A fresh subscribe supersedes any earlier drop of this channel; + // clearing the mark keeps the harness from re-clearing state it has + // already rebuilt. + if let Ok(mut dropped) = state.dropped_channels.lock() { + dropped.remove(&channel_id); + } state .active_subscriptions .insert(channel_id, channel_sub_id(channel_id)); @@ -1554,8 +1625,9 @@ async fn run_background_task( relay_url: String, agent_pubkey_hex: String, auth_tag: Option, + dropped_channels: DroppedChannels, ) { - let mut state = BgState::new(); + let mut state = BgState::with_dropped_channels(dropped_channels); let handshake_ok = process_handshake_buffer( &mut ws, @@ -2711,6 +2783,7 @@ async fn drain_rate_limited_pending( None => { warn!("missing filter for channel {channel_id} in rate_limited_pending — dropping"); state.rate_limited_pending.remove(&channel_id); + state.record_channel_drop(channel_id, "missing filter in rate_limited_pending"); continue; } }; @@ -2768,6 +2841,7 @@ async fn drain_resubscribe_retry( None => { warn!("missing filter for channel {channel_id} in resubscribe_retry — dropping"); state.resubscribe_retry.remove(&channel_id); + state.record_channel_drop(channel_id, "missing filter in resubscribe_retry"); continue; } }; @@ -3537,6 +3611,7 @@ fn drop_channel_on_access_denied(state: &mut BgState, sub_id: &str, message: &st ); state.active_subscriptions.remove(&channel_id); state.clear_channel_state(&channel_id); + state.record_channel_drop(channel_id, "access denied"); true } @@ -5053,6 +5128,82 @@ mod tests { ); } + #[test] + fn access_denied_drop_is_reported_to_the_harness() { + let sink: DroppedChannels = DroppedChannels::default(); + let mut state = BgState::with_dropped_channels(sink.clone()); + let channel_id = Uuid::new_v4(); + subscribe_channel(&mut state, channel_id); + + drop_channel_on_access_denied( + &mut state, + &channel_sub_id(channel_id), + "restricted: not a channel member", + ); + + assert!( + sink.lock().unwrap().contains(&channel_id), + "a dropped channel must be reported so the harness can re-subscribe it" + ); + } + + #[test] + fn a_kept_channel_is_never_reported_as_dropped() { + let sink: DroppedChannels = DroppedChannels::default(); + let mut state = BgState::with_dropped_channels(sink.clone()); + let channel_id = Uuid::new_v4(); + subscribe_channel(&mut state, channel_id); + + // Connection-level denial: reconnect handles it, the channel survives. + drop_channel_on_access_denied( + &mut state, + &channel_sub_id(channel_id), + "restricted: insufficient scope", + ); + + assert!( + sink.lock().unwrap().is_empty(), + "only permanent per-channel drops may be reported" + ); + } + + #[test] + fn resubscribing_a_channel_clears_its_drop_report() { + let sink: DroppedChannels = DroppedChannels::default(); + let mut state = BgState::with_dropped_channels(sink.clone()); + let channel_id = Uuid::new_v4(); + subscribe_channel(&mut state, channel_id); + drop_channel_on_access_denied( + &mut state, + &channel_sub_id(channel_id), + "restricted: channel access revoked", + ); + assert!(sink.lock().unwrap().contains(&channel_id)); + + // The harness re-subscribes after the membership notification. + subscribe_channel(&mut state, channel_id); + + assert!( + sink.lock().unwrap().is_empty(), + "a fresh subscribe supersedes the stale drop report" + ); + assert!(state.active_filters.contains_key(&channel_id)); + } + + #[test] + fn missing_filter_drop_paths_report_the_channel() { + let sink: DroppedChannels = DroppedChannels::default(); + let mut state = BgState::with_dropped_channels(sink.clone()); + let channel_id = Uuid::new_v4(); + + state.record_channel_drop(channel_id, "missing filter in resubscribe_retry"); + + assert!( + sink.lock().unwrap().contains(&channel_id), + "the missing-filter drop paths must also reach the harness" + ); + } + #[test] fn dropped_channel_is_not_resubscribed_so_loop_cannot_re_form() { let mut state = BgState::new();