fix: WebSocket hardening + agent reliability (8 WS bugs, member-only discovery, mention tag fix) (#67)

crates/sprout-acp/README.md

@@ -40,7 +40,7 @@ This prints an `nsec1...` private key and an API token. **Save both immediately
 
 The harness discovers channels by querying the relay with the agent's authenticated identity.
 
-**Open channels** (the default for local dev) are accessible to any authenticated pubkey — no extra setup needed. Just start the harness and it will find and subscribe to all open channels.
+By default, the harness discovers only channels the agent is a **member** of (`GET /api/channels?member=true`). When the agent is added to a new channel, the membership notification subscription auto-subscribes to it.
 
 **Private channels** require explicit membership. The relay doesn't yet have a REST/event API for managing channel members — this is a known gap. For now, use `create_channel` via the Sprout MCP tools to create new channels (the creator is automatically a member).
 

crates/sprout-acp/src/config.rs

@@ -142,7 +142,7 @@ pub struct CliArgs {
     #[arg(long, env = "SPROUT_ACP_CONFIG", default_value = "./sprout-acp.toml")]
     pub config: PathBuf,
 
-    #[arg(long, env = "SPROUT_ACP_DEDUP", default_value = "drop", value_enum)]
+    #[arg(long, env = "SPROUT_ACP_DEDUP", default_value = "queue", value_enum)]
     pub dedup: DedupMode,
 
     #[arg(long, env = "SPROUT_ACP_NO_IGNORE_SELF")]
@@ -566,7 +566,7 @@ mod tests {
             system_prompt: None,
             initial_message: None,
             subscribe_mode: mode,
-            dedup_mode: DedupMode::Drop,
+            dedup_mode: DedupMode::Queue,
             ignore_self: true,
             kinds_override: None,
             channels_override: None,

crates/sprout-acp/src/relay.rs

@@ -24,16 +24,28 @@ use std::time::Duration;
 
 // ─── Named constants ──────────────────────────────────────────────────────────
 
-/// Capacity of the event channel from background task to harness.
-const EVENT_CHANNEL_CAPACITY: usize = 256;
+/// Default capacity of the event channel from background task to harness.
+/// Override with `SPROUT_ACP_EVENT_BUFFER` env var at startup.
+const EVENT_CHANNEL_CAPACITY_DEFAULT: usize = 256;
 /// Capacity of the command channel from harness to background task.
 const CMD_CHANNEL_CAPACITY: usize = 64;
+
+/// Read the event channel capacity from the environment, falling back to the
+/// compiled-in default. Parsed once at call-site (connect time).
+fn event_channel_capacity() -> usize {
+    std::env::var("SPROUT_ACP_EVENT_BUFFER")
+        .ok()
+        .and_then(|v| v.parse::<usize>().ok())
+        .unwrap_or(EVENT_CHANNEL_CAPACITY_DEFAULT)
+}
 /// Maximum number of seen event IDs before the dedup set is cleared.
 const SEEN_ID_LIMIT: usize = 12_000;
 /// Seconds subtracted from `since` on resubscribe to tolerate clock skew.
 const SINCE_SKEW_SECS: u64 = 5;
 /// Timeout for the NIP-42 auth handshake steps.
 const AUTH_TIMEOUT: Duration = Duration::from_secs(5);
+/// Timeout for the TCP + WebSocket handshake in `do_connect`.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
 
 use futures_util::{SinkExt, StreamExt};
 use nostr::{Event, EventBuilder, Keys, Kind, Tag, Url as NostrUrl};
@@ -190,7 +202,7 @@ impl HarnessRelay {
         // Perform the initial connection and auth handshake.
         let (ws, _buffer) = do_connect(relay_url, keys, api_token).await?;
 
-        let (event_tx, event_rx) = mpsc::channel::<Option<SproutEvent>>(EVENT_CHANNEL_CAPACITY);
+        let (event_tx, event_rx) = mpsc::channel::<Option<SproutEvent>>(event_channel_capacity());
         let (cmd_tx, cmd_rx) = mpsc::channel::<RelayCommand>(CMD_CHANNEL_CAPACITY);
 
         let bg_keys = keys.clone();
@@ -227,10 +239,10 @@ impl HarnessRelay {
         })
     }
 
-    /// Discover channels the harness is a member of via `GET /api/channels`.
+    /// Discover channels the agent is a member of via `GET /api/channels?member=true`.
     pub async fn discover_channels(&self) -> Result<Vec<Uuid>, RelayError> {
         let http_url = relay_ws_to_http(&self.relay_url);
-        let url = format!("{http_url}/api/channels");
+        let url = format!("{http_url}/api/channels?member=true");
 
         let builder = self.http.get(&url);
         let builder = apply_auth(builder, &self.api_token, &self.keys);
@@ -359,6 +371,11 @@ struct BgState {
     membership_last_seen: Option<u64>,
     /// Whether the membership notification subscription is active.
     membership_sub_active: bool,
+    /// Oldest dropped channel-event timestamp per channel, keyed by channel_id.
+    /// Mirrors `membership_dropped_since` but for ordinary channel events.
+    /// On reconnect resubscribe, `since` = min(last_seen, channel_dropped_since).
+    /// Cleared per-channel after a successful resubscribe.
+    channel_dropped_since: HashMap<Uuid, u64>,
 }
 
 impl BgState {
@@ -371,6 +388,7 @@ impl BgState {
             membership_dropped_since: None,
             membership_last_seen: None,
             membership_sub_active: false,
+            channel_dropped_since: HashMap::new(),
         }
     }
 
@@ -423,9 +441,10 @@ async fn run_background_task(
         tokio::select! {
             // ── Incoming WebSocket message ────────────────────────────────────
             raw = ws.next() => {
-                match raw {
+                // Determine if the socket is lost.
+                let socket_lost = match raw {
                     Some(Ok(msg)) => {
-                        if !handle_ws_message(
+                        !handle_ws_message(
                             msg,
                             &mut ws,
                             &event_tx,
@@ -435,41 +454,62 @@ async fn run_background_task(
                             api_token.as_deref(),
                         )
                         .await
-                        {
-                            // handle_ws_message returns false on connection loss.
-                            // Signal the caller and wait for a Reconnect command.
-                            let _ = event_tx.send(None).await;
-                            wait_for_reconnect(
-                                &mut ws,
-                                &mut cmd_rx,
-                                &mut state,
-                                &keys,
-                                &relay_url,
-                                api_token.as_deref(),
-                                &agent_pubkey_hex,
-                                false, // skip_drain: wait for Reconnect command
-                            )
-                            .await;
-                        }
                     }
                     Some(Err(e)) => {
                         warn!("WebSocket error in background task: {e}");
-                        let _ = event_tx.send(None).await;
-                        wait_for_reconnect(
-                            &mut ws,
-                            &mut cmd_rx,
-                            &mut state,
-                            &keys,
-                            &relay_url,
-                            api_token.as_deref(),
-                            &agent_pubkey_hex,
-                            false,
-                        )
-                        .await;
+                        true
                     }
                     None => {
                         debug!("WebSocket stream ended");
-                        let _ = event_tx.send(None).await;
+                        true
+                    }
+                };
+
+                if socket_lost {
+                    // Signal the caller, then attempt autonomous reconnect.
+                    let _ = event_tx.send(None).await;
+                    let reconnected = try_autonomous_reconnect(
+                        &mut ws,
+                        &mut state,
+                        &keys,
+                        &relay_url,
+                        api_token.as_deref(),
+                        &agent_pubkey_hex,
+                    )
+                    .await;
+                    if reconnected {
+                        // Drain stale Reconnect commands the caller queued after
+                        // seeing the None sentinel. Process other commands normally
+                        // so Subscribe/Unsubscribe/SubscribeMembership aren't lost.
+                        while let Ok(cmd) = cmd_rx.try_recv() {
+                            match cmd {
+                                RelayCommand::Reconnect => {
+                                    debug!("drained stale Reconnect command after autonomous reconnect");
+                                }
+                                RelayCommand::Shutdown => {
+                                    debug!("shutdown received during reconnect drain");
+                                    return;
+                                }
+                                RelayCommand::Subscribe { channel_id, filter } => {
+                                    send_subscribe(&mut ws, &state, channel_id, &agent_pubkey_hex, None, &filter).await;
+                                    state.active_subscriptions.insert(channel_id, channel_sub_id(channel_id));
+                                    state.active_filters.insert(channel_id, filter);
+                                }
+                                RelayCommand::Unsubscribe { channel_id } => {
+                                    if let Some(sub_id) = state.active_subscriptions.remove(&channel_id) {
+                                        let msg = json!(["CLOSE", sub_id]);
+                                        if let Ok(text) = serde_json::to_string(&msg) {
+                                            let _ = ws.send(Message::Text(text.into())).await;
+                                        }
+                                    }
+                                }
+                                RelayCommand::SubscribeMembership => {
+                                    let _ = send_membership_subscribe(&mut ws, &agent_pubkey_hex, None).await;
+                                    state.membership_sub_active = true;
+                                }
+                            }
+                        }
+                    } else {
                         wait_for_reconnect(
                             &mut ws,
                             &mut cmd_rx,
@@ -603,6 +643,8 @@ async fn handle_ws_message(
                             Err(mpsc::error::TrySendError::Closed(_)) => return false,
                         }
                     } else if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) {
+                        let ts = event.created_at.as_u64();
+                        let event_id_hex = event.id.to_hex();
                         if state.record_event(channel_id, &event) {
                             let sprout_event = SproutEvent {
                                 channel_id,
@@ -611,7 +653,21 @@ async fn handle_ws_message(
                             match event_tx.try_send(Some(sprout_event)) {
                                 Ok(()) => {}
                                 Err(mpsc::error::TrySendError::Full(_)) => {
-                                    warn!("event channel full — dropping event for channel {channel_id}");
+                                    // Remove from dedup set so the replayed event
+                                    // won't be rejected as a duplicate after reconnect.
+                                    state.seen_ids.remove(&event_id_hex);
+                                    // Track the oldest dropped timestamp so reconnect
+                                    // replay starts early enough to re-deliver it.
+                                    state
+                                        .channel_dropped_since
+                                        .entry(channel_id)
+                                        .and_modify(|d| *d = (*d).min(ts))
+                                        .or_insert(ts);
+                                    warn!(
+                                        channel_id = %channel_id,
+                                        ts,
+                                        "event channel full — dropping event for channel {channel_id} — will replay from {ts} on reconnect"
+                                    );
                                 }
                                 Err(mpsc::error::TrySendError::Closed(_)) => {
                                     // Receiver dropped — shut down.
@@ -673,6 +729,115 @@ async fn handle_ws_message(
     }
 }
 
+/// Resubscribe all active channels and membership notifications after a
+/// successful reconnect. Computes `since = min(last_seen, channel_dropped_since)`
+/// per channel, and only clears the drop tracker when the REQ is confirmed sent.
+async fn resubscribe_after_reconnect(
+    ws: &mut WsStream,
+    state: &mut BgState,
+    agent_pubkey_hex: &str,
+) {
+    let channels: Vec<Uuid> = state.active_subscriptions.keys().copied().collect();
+    if !channels.is_empty() {
+        info!(
+            "resubscribing to {} channel(s) after reconnect",
+            channels.len()
+        );
+        for channel_id in channels {
+            let last_seen = state.last_seen.get(&channel_id).copied();
+            let dropped = state.channel_dropped_since.get(&channel_id).copied();
+            let since = match (last_seen, dropped) {
+                (Some(l), Some(d)) => Some(l.min(d)),
+                (Some(l), None) => Some(l),
+                (None, Some(d)) => Some(d),
+                (None, None) => None,
+            };
+            let filter = state
+                .active_filters
+                .get(&channel_id)
+                .cloned()
+                .unwrap_or(ChannelFilter {
+                    kinds: None,
+                    require_mention: false,
+                });
+            let sent =
+                send_subscribe(ws, state, channel_id, agent_pubkey_hex, since, &filter).await;
+            if sent {
+                state.channel_dropped_since.remove(&channel_id);
+            }
+        }
+    }
+
+    if state.membership_sub_active {
+        let replay_since = match (state.membership_dropped_since, state.membership_last_seen) {
+            (Some(d), Some(l)) => Some(d.min(l)),
+            (Some(d), None) => Some(d),
+            (None, Some(l)) => Some(l),
+            (None, None) => None,
+        };
+        let sent = send_membership_subscribe(ws, agent_pubkey_hex, replay_since).await;
+        if sent {
+            state.membership_dropped_since = None;
+        }
+    }
+}
+
+/// Attempt autonomous reconnect on socket loss — 3 attempts with 1s→2s→4s backoff.
+///
+/// If any attempt succeeds, resubscribes all active channels and membership
+/// notifications via [`resubscribe_after_reconnect`] and returns `true`.
+/// If all 3 attempts fail, returns `false` so the caller can fall back to
+/// `wait_for_reconnect` (which blocks until the caller sends a `Reconnect`).
+#[allow(clippy::too_many_arguments)]
+async fn try_autonomous_reconnect(
+    ws: &mut WsStream,
+    state: &mut BgState,
+    keys: &Keys,
+    relay_url: &str,
+    api_token: Option<&str>,
+    agent_pubkey_hex: &str,
+) -> bool {
+    let backoffs = [
+        Duration::from_secs(1),
+        Duration::from_secs(2),
+        Duration::from_secs(4),
+    ];
+
+    for (attempt, delay) in backoffs.iter().enumerate() {
+        info!(
+            "autonomous reconnect attempt {}/{} to {relay_url}…",
+            attempt + 1,
+            backoffs.len()
+        );
+        match do_connect(relay_url, keys, api_token).await {
+            Ok((new_ws, _buffer)) => {
+                *ws = new_ws;
+                info!("autonomous reconnect succeeded (attempt {})", attempt + 1);
+                resubscribe_after_reconnect(ws, state, agent_pubkey_hex).await;
+                return true;
+            }
+            Err(e) => {
+                warn!(
+                    "autonomous reconnect attempt {} failed: {e} — {}",
+                    attempt + 1,
+                    if attempt + 1 < backoffs.len() {
+                        format!("retrying in {}s", delay.as_secs())
+                    } else {
+                        "falling back to caller-driven reconnect".to_string()
+                    }
+                );
+                // Skip sleep on the final attempt — no point waiting before
+                // falling back to caller-driven reconnect.
+                if attempt + 1 < backoffs.len() {
+                    tokio::time::sleep(*delay).await;
+                }
+            }
+        }
+    }
+
+    false
+}
+
 /// Attempt reconnection with exponential backoff. Resubscribes all active
 /// channels with `since` filters on success.
 ///
@@ -725,42 +890,7 @@ async fn wait_for_reconnect(
             Ok((new_ws, _buffer)) => {
                 *ws = new_ws;
                 info!("relay reconnected to {relay_url}");
-
-                // Resubscribe all active channels with `since` filter.
-                let channels: Vec<Uuid> = state.active_subscriptions.keys().copied().collect();
-                if !channels.is_empty() {
-                    info!("resubscribing to {} channel(s)", channels.len());
-                    for channel_id in channels {
-                        let since = state.last_seen.get(&channel_id).copied();
-                        let filter = state.active_filters.get(&channel_id).cloned().unwrap_or(
-                            ChannelFilter {
-                                kinds: None,
-                                require_mention: false,
-                            },
-                        );
-                        send_subscribe(ws, state, channel_id, agent_pubkey_hex, since, &filter)
-                            .await;
-                    }
-                }
-
-                // Resubscribe to membership notifications if active.
-                // Use the oldest of (dropped, last_seen) so dropped events are replayed.
-                if state.membership_sub_active {
-                    let replay_since =
-                        match (state.membership_dropped_since, state.membership_last_seen) {
-                            (Some(d), Some(l)) => Some(d.min(l)),
-                            (Some(d), None) => Some(d),
-                            (None, Some(l)) => Some(l),
-                            (None, None) => None,
-                        };
-                    let sent = send_membership_subscribe(ws, agent_pubkey_hex, replay_since).await;
-                    if sent {
-                        // Only clear drop tracker if the REQ was actually sent —
-                        // if send failed, retain it so the next reconnect retries.
-                        state.membership_dropped_since = None;
-                    }
-                }
-
+                resubscribe_after_reconnect(ws, state, agent_pubkey_hex).await;
                 return;
             }
             Err(e) => {
@@ -782,6 +912,8 @@ async fn wait_for_reconnect(
 /// - `#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`].
+///
+/// Returns `true` if the REQ was successfully written to the WebSocket.
 async fn send_subscribe(
     ws: &mut WsStream,
     _state: &BgState,
@@ -789,7 +921,7 @@ async fn send_subscribe(
     agent_pubkey_hex: &str,
     since: Option<u64>,
     filter: &ChannelFilter,
-) {
+) -> bool {
     let sub_id = channel_sub_id(channel_id);
 
     let mut req_filter = serde_json::Map::new();
@@ -821,10 +953,8 @@ async fn send_subscribe(
     let req = json!(["REQ", sub_id, Value::Object(req_filter)]);
 
     match serde_json::to_string(&req) {
-        Ok(text) => {
-            if let Err(e) = ws.send(Message::Text(text.into())).await {
-                warn!("failed to send REQ for channel {channel_id}: {e}");
-            } else {
+        Ok(text) => match ws.send(Message::Text(text.into())).await {
+            Ok(()) => {
                 debug!(
                     "subscribed to channel {channel_id}{}",
                     if since.is_some() {
@@ -833,10 +963,16 @@ async fn send_subscribe(
                         " (since=now)"
                     }
                 );
+                true
             }
-        }
+            Err(e) => {
+                warn!("failed to send REQ for channel {channel_id}: {e}");
+                false
+            }
+        },
         Err(e) => {
             warn!("failed to serialize REQ for channel {channel_id}: {e}");
+            false
         }
     }
 }
@@ -1078,8 +1214,9 @@ async fn do_connect(
         .parse::<url::Url>()
         .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?;
 
-    let (ws, _response) = connect_async(parsed.as_str())
+    let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str()))
         .await
+        .map_err(|_| RelayError::ConnectionClosed)? // timeout → treat as connection failure
         .map_err(|e| RelayError::WebSocket(Box::new(e)))?;
     debug!("connected to relay at {relay_url}");
 
@@ -1648,4 +1785,99 @@ mod tests {
             state.seen_ids.len()
         );
     }
+
+    // ── Bug 5: channel_dropped_since tracking ─────────────────────────────────
+
+    /// Test 8: channel_dropped_since records the OLDEST dropped timestamp.
+    ///
+    /// Simulates the backpressure path directly on BgState:
+    /// - First drop at ts=1000 → entry is 1000
+    /// - Second drop at ts=2000 (later) → entry stays 1000 (min)
+    /// - Third drop at ts=500 (earlier) → entry updates to 500 (min)
+    #[test]
+    fn acp_records_channel_dropped_since_on_backpressure() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+
+        // Simulate the backpressure path: record ts=1000.
+        let ts1: u64 = 1_000;
+        state
+            .channel_dropped_since
+            .entry(channel_id)
+            .and_modify(|d| *d = (*d).min(ts1))
+            .or_insert(ts1);
+        assert_eq!(
+            state.channel_dropped_since.get(&channel_id).copied(),
+            Some(1_000),
+            "first drop should record ts=1000"
+        );
+
+        // Later timestamp (2000) — entry should stay at 1000.
+        let ts2: u64 = 2_000;
+        state
+            .channel_dropped_since
+            .entry(channel_id)
+            .and_modify(|d| *d = (*d).min(ts2))
+            .or_insert(ts2);
+        assert_eq!(
+            state.channel_dropped_since.get(&channel_id).copied(),
+            Some(1_000),
+            "later drop should not overwrite earlier timestamp"
+        );
+
+        // Earlier timestamp (500) — entry should update to 500.
+        let ts3: u64 = 500;
+        state
+            .channel_dropped_since
+            .entry(channel_id)
+            .and_modify(|d| *d = (*d).min(ts3))
+            .or_insert(ts3);
+        assert_eq!(
+            state.channel_dropped_since.get(&channel_id).copied(),
+            Some(500),
+            "earlier drop should update entry to 500"
+        );
+    }
+
+    /// Test 9: reconnect since filter = min(last_seen, channel_dropped_since) - SINCE_SKEW_SECS.
+    ///
+    /// With last_seen=1000 and channel_dropped_since=900, the effective since
+    /// passed to send_subscribe should be min(1000, 900) - SINCE_SKEW_SECS = 895.
+    #[test]
+    fn acp_reconnect_uses_dropped_since_for_replay() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+
+        // Set up state: last_seen=1000, channel_dropped_since=900.
+        state.last_seen.insert(channel_id, 1_000);
+        state.channel_dropped_since.insert(channel_id, 900);
+
+        // Compute the since value the reconnect path would use.
+        let last_seen = state.last_seen.get(&channel_id).copied();
+        let dropped = state.channel_dropped_since.get(&channel_id).copied();
+        let since = match (last_seen, dropped) {
+            (Some(l), Some(d)) => Some(l.min(d)),
+            (Some(l), None) => Some(l),
+            (None, Some(d)) => Some(d),
+            (None, None) => None,
+        };
+
+        // The since passed to send_subscribe (which subtracts SINCE_SKEW_SECS internally).
+        assert_eq!(since, Some(900), "since should be min(1000, 900) = 900");
+
+        // After subtracting skew (as send_subscribe does), the REQ filter value is:
+        let req_since = since.unwrap().saturating_sub(SINCE_SKEW_SECS);
+        assert_eq!(
+            req_since, 895,
+            "REQ since filter should be 900 - {} = 895",
+            SINCE_SKEW_SECS
+        );
+
+        // Simulate clearing after resubscribe.
+        state.channel_dropped_since.remove(&channel_id);
+        assert!(
+            !state.channel_dropped_since.contains_key(&channel_id),
+            "channel_dropped_since should be cleared after resubscribe"
+        );
+    }
 }

crates/sprout-db/src/channel.rs

@@ -707,8 +707,20 @@ pub async fn get_accessible_channels(
     pool: &MySqlPool,
     pubkey: &[u8],
     visibility_filter: Option<&str>,
+    member_only: Option<bool>,
 ) -> Result<Vec<AccessibleChannel>> {
-    let base = r#"
+    // When `member_only` is `Some(true)`, restrict to channels where the user
+    // has an active membership (cm.channel_id IS NOT NULL). This is a strict
+    // subset of the default result set and is pushed into SQL so the LIMIT 1000
+    // applies to the filtered set, not the pre-filter set.
+    let membership_clause = if member_only == Some(true) {
+        "AND cm.channel_id IS NOT NULL"
+    } else {
+        "AND (c.visibility = 'open' OR cm.channel_id IS NOT NULL)"
+    };
+
+    let base = format!(
+        r#"
         SELECT DISTINCT c.id, c.name, c.channel_type, c.visibility, c.description, c.canvas,
                c.created_by, c.created_at, c.updated_at, c.archived_at, c.deleted_at,
                c.nip29_group_id, c.topic_required, c.max_members,
@@ -719,8 +731,9 @@ pub async fn get_accessible_channels(
         LEFT JOIN channel_members cm
             ON c.id = cm.channel_id AND cm.pubkey = ? AND cm.removed_at IS NULL
         WHERE c.deleted_at IS NULL
-          AND (c.visibility = 'open' OR cm.channel_id IS NOT NULL)
-    "#;
+          {membership_clause}
+    "#
+    );
 
     let sql = if visibility_filter.is_some() {
         format!("{base}  AND c.visibility = ?\n        ORDER BY FIELD(c.channel_type, 'stream', 'forum', 'dm'), c.name\n        LIMIT 1000")

crates/sprout-db/src/lib.rs

@@ -403,12 +403,14 @@ impl Db {
     ///
     /// If `visibility_filter` is `Some("open")` or `Some("private")`, only channels
     /// with that visibility are returned.
+    /// If `member_only` is `Some(true)`, only channels the user is a member of are returned.
     pub async fn get_accessible_channels(
         &self,
         pubkey: &[u8],
         visibility_filter: Option<&str>,
+        member_only: Option<bool>,
     ) -> Result<Vec<channel::AccessibleChannel>> {
-        channel::get_accessible_channels(&self.pool, pubkey, visibility_filter).await
+        channel::get_accessible_channels(&self.pool, pubkey, visibility_filter, member_only).await
     }
 
     /// Returns all bot-role members with aggregated channel names.

crates/sprout-mcp/Cargo.toml

@@ -41,3 +41,7 @@ tracing-subscriber = { workspace = true }
 thiserror = { workspace = true }
 anyhow = { workspace = true }
 url = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["test-util"] }
+tokio-tungstenite = { workspace = true }

crates/sprout-mcp/src/lib.rs

@@ -37,11 +37,27 @@
 //! `SPROUT_API_TOKEN` is set the token is embedded in the auth event tags so the relay can
 //! verify the agent's API permissions.
 //!
-//! ## WebSocket Reconnection
+//! ## WebSocket Connection Management
 //!
-//! [`relay_client::RelayClient`] supports automatic reconnection with exponential backoff
-//! (1 s → 2 s → 4 s → … → 30 s cap). After reconnecting it re-authenticates via NIP-42 and
-//! resubmits all subscriptions that were active at the time of the disconnect.
+//! [`relay_client::RelayClient`] uses a background tokio task that owns the WebSocket
+//! connection. The background task:
+//!
+//! - Responds to Ping frames immediately — preventing relay disconnects during long LLM turns
+//! - Handles mid-session NIP-42 AUTH challenges automatically
+//! - Reconnects with exponential backoff (1 s → 2 s → 4 s → … → 30 s cap) on any
+//!   connection loss, without any action required from the caller
+//! - Re-authenticates via NIP-42 after each reconnect
+//! - Replays all active subscriptions after reconnect
+//!
+//! ```text
+//! RelayClient (Clone)
+//!   ├── cmd_tx: mpsc::Sender<RelayCommand>   ← send_event / subscribe / close
+//!   └── bg_handle: JoinHandle<()>
+//!         └── run_background_task()
+//!               ├── ws.next()  → handle_ws_message()   // Ping→Pong, AUTH→respond, OK→resolve
+//!               ├── cmd_rx     → handle_command()       // SendEvent, Subscribe, Close
+//!               └── tick       → expire_timed_out()     // 10s timeouts
+//! ```
 //!
 //! ## Available Tools
 //!

crates/sprout-relay/src/api/agents.rs

@@ -30,7 +30,7 @@ pub async fn agents_handler(
     // Get requester's accessible channels to filter bot channel visibility.
     let accessible_channels = state
         .db
-        .get_accessible_channels(&pubkey_bytes, None)
+        .get_accessible_channels(&pubkey_bytes, None, None)
         .await
         .map_err(|e| {
             tracing::error!("agents: failed to load accessible channels: {e}");

crates/sprout-relay/src/api/channels.rs

@@ -26,6 +26,8 @@ use super::{api_error, extract_auth_context, internal_error};
 pub struct ListChannelsParams {
     /// Optional visibility filter: `"open"` or `"private"`.
     pub visibility: Option<String>,
+    /// When `true`, return only channels the user is a member of.
+    pub member: Option<bool>,
 }
 
 /// Returns all channels accessible to the authenticated user.
@@ -43,7 +45,7 @@ pub async fn channels_handler(
 
     let channels = state
         .db
-        .get_accessible_channels(&pubkey_bytes, params.visibility.as_deref())
+        .get_accessible_channels(&pubkey_bytes, params.visibility.as_deref(), params.member)
         .await
         .map_err(|e| internal_error(&format!("db error: {e}")))?;
 

crates/sprout-relay/src/connection.rs

@@ -20,6 +20,10 @@ use crate::handlers;
 use crate::protocol::{ClientMessage, RelayMessage};
 use crate::state::AppState;
 
+/// Number of buffer-full events tolerated before cancelling a slow client.
+/// Prevents transient read stalls from hard-disconnecting agents mid-inference.
+pub(crate) const SLOW_CLIENT_GRACE_LIMIT: u8 = 3;
+
 /// NIP-42 authentication state for a single connection.
 #[derive(Debug, Clone)]
 pub enum AuthState {
@@ -37,7 +41,7 @@ pub enum AuthState {
 /// Per-connection state split by access pattern:
 /// - `auth_state`: RwLock (read-heavy after initial auth)
 /// - `subscriptions`: Mutex (write-heavy during REQ/CLOSE)
-/// - `send_tx`, `cancel`: outside any lock (Clone+Send, no coordination needed)
+/// - `send_tx`, `ctrl_tx`, `cancel`: outside any lock (Clone+Send, no coordination needed)
 pub struct ConnectionState {
     /// Unique identifier for this connection.
     pub conn_id: Uuid,
@@ -47,23 +51,41 @@ pub struct ConnectionState {
     pub auth_state: RwLock<AuthState>,
     /// Active subscriptions keyed by subscription ID.
     pub subscriptions: Mutex<HashMap<String, Vec<Filter>>>,
-    /// Sender for outbound WebSocket messages.
+    /// Sender for outbound data messages (EVENT, NOTICE, OK, etc.).
     pub send_tx: mpsc::Sender<WsMessage>,
+    /// Sender for outbound control frames (Pong, Close).
+    /// Separate channel with priority drain — if this channel fills too,
+    /// the connection is closed (writer is completely stalled).
+    pub ctrl_tx: mpsc::Sender<WsMessage>,
     /// Token used to signal graceful shutdown of this connection's tasks.
     pub cancel: CancellationToken,
+    /// Consecutive buffer-full events. Cancel only after [`SLOW_CLIENT_GRACE_LIMIT`].
+    /// Shared with `ConnectionManager::ConnEntry` so both direct sends and
+    /// fan-out broadcasts track the same counter.
+    pub backpressure_count: Arc<AtomicU8>,
 }
 
 impl ConnectionState {
-    /// Sends a message to this connection's outbound channel.
+    /// Sends a data message to this connection's outbound channel.
     ///
-    /// If the send buffer is full (slow client), cancels the connection
-    /// via the `CancellationToken` to prevent unbounded memory growth.
+    /// On a full buffer, increments the backpressure counter. The first
+    /// [`SLOW_CLIENT_GRACE_LIMIT`] occurrences log a warning; sustained
+    /// backpressure cancels the connection to prevent unbounded memory growth.
     pub fn send(&self, msg: String) -> bool {
         match self.send_tx.try_send(WsMessage::Text(msg.into())) {
-            Ok(_) => true,
+            Ok(_) => {
+                // Successful send resets the grace counter.
+                self.backpressure_count.store(0, Ordering::Relaxed);
+                true
+            }
             Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
-                warn!(conn_id = %self.conn_id, "send buffer full — closing slow client");
-                self.cancel.cancel();
+                let count = self.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1;
+                if count >= SLOW_CLIENT_GRACE_LIMIT {
+                    warn!(conn_id = %self.conn_id, count, "sustained backpressure — closing slow client");
+                    self.cancel.cancel();
+                } else {
+                    warn!(conn_id = %self.conn_id, count, grace = SLOW_CLIENT_GRACE_LIMIT, "send buffer full — grace {count}/{SLOW_CLIENT_GRACE_LIMIT}");
+                }
                 false
             }
             Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
@@ -92,6 +114,11 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
     let cancel = CancellationToken::new();
 
     let (tx, rx) = mpsc::channel::<WsMessage>(state.config.send_buffer_size);
+    // Control channel for Pong/Close — small capacity, guaranteed delivery
+    // even when the data buffer is full.
+    let (ctrl_tx, ctrl_rx) = mpsc::channel::<WsMessage>(8);
+
+    let backpressure_count = Arc::new(AtomicU8::new(0));
 
     let conn = Arc::new(ConnectionState {
         conn_id,
@@ -101,7 +128,9 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
         }),
         subscriptions: Mutex::new(HashMap::new()),
         send_tx: tx.clone(),
+        ctrl_tx: ctrl_tx.clone(),
         cancel: cancel.clone(),
+        backpressure_count: Arc::clone(&backpressure_count),
     });
 
     info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established");
@@ -117,19 +146,22 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
     }
 
     // Register after challenge succeeds — avoids leaked entries on early disconnect.
-    state
-        .conn_manager
-        .register(conn_id, tx.clone(), cancel.clone());
+    state.conn_manager.register(
+        conn_id,
+        tx.clone(),
+        cancel.clone(),
+        Arc::clone(&backpressure_count),
+    );
 
     let (ws_send, ws_recv) = socket.split();
 
     let send_cancel = cancel.child_token();
-    let send_task = tokio::spawn(send_loop(ws_send, rx, send_cancel));
+    let send_task = tokio::spawn(send_loop(ws_send, rx, ctrl_rx, send_cancel));
 
     let missed_pongs = Arc::new(AtomicU8::new(0));
     let heartbeat_cancel = cancel.clone();
     let heartbeat_task = tokio::spawn(heartbeat_loop(
-        tx.clone(),
+        ctrl_tx,
         Arc::clone(&missed_pongs),
         heartbeat_cancel,
     ));
@@ -154,29 +186,55 @@ pub async fn handle_connection(socket: WebSocket, state: Arc<AppState>, addr: So
     drop(permit);
 }
 
+/// Outbound send loop with control-frame priority.
+///
+/// Control frames (Pong, Close) are drained first on every iteration,
+/// giving them priority over data frames. If the underlying socket writer
+/// is stalled, control frames queue in the small ctrl_rx buffer; callers
+/// treat a full control channel as terminal (Bug 7 fix).
 async fn send_loop(
     mut ws_send: futures_util::stream::SplitSink<WebSocket, WsMessage>,
-    mut rx: mpsc::Receiver<WsMessage>,
+    mut data_rx: mpsc::Receiver<WsMessage>,
+    mut ctrl_rx: mpsc::Receiver<WsMessage>,
     cancel: CancellationToken,
 ) {
     loop {
-        tokio::select! {
-            Some(msg) = rx.recv() => {
-                if ws_send.send(msg).await.is_err() {
-                    break;
-                }
+        // Priority: drain all pending control frames before data.
+        while let Ok(ctrl_msg) = ctrl_rx.try_recv() {
+            if ws_send.send(ctrl_msg).await.is_err() {
+                return;
             }
+        }
+
+        tokio::select! {
+            // Biased: cancel > control > data. Cancel must win immediately
+            // so backpressure-triggered shutdown isn't starved by queued data.
+            biased;
             _ = cancel.cancelled() => {
                 let _ = ws_send.send(WsMessage::Close(None)).await;
                 break;
             }
+            Some(ctrl_msg) = ctrl_rx.recv() => {
+                if ws_send.send(ctrl_msg).await.is_err() {
+                    break;
+                }
+            }
+            Some(msg) = data_rx.recv() => {
+                if ws_send.send(msg).await.is_err() {
+                    break;
+                }
+            }
         }
     }
 }
 
 /// 3 missed pongs → disconnect.
+///
+/// Sends Ping through the control channel so it isn't blocked by a full
+/// data buffer. Uses `try_send` to keep the select loop responsive to
+/// cancellation — a full control channel means the writer is stalled.
 async fn heartbeat_loop(
-    tx: mpsc::Sender<WsMessage>,
+    ctrl_tx: mpsc::Sender<WsMessage>,
     missed_pongs: Arc<AtomicU8>,
     cancel: CancellationToken,
 ) {
@@ -194,7 +252,9 @@ async fn heartbeat_loop(
                     cancel.cancel();
                     break;
                 }
-                if tx.send(WsMessage::Ping(axum::body::Bytes::new())).await.is_err() {
+                if ctrl_tx.try_send(WsMessage::Ping(axum::body::Bytes::new())).is_err() {
+                    warn!("control channel full — cannot send Ping, closing");
+                    cancel.cancel();
                     break;
                 }
             }
@@ -241,7 +301,14 @@ async fn recv_loop(
                         missed_pongs.store(0, Ordering::Relaxed);
                     }
                     Some(Ok(WsMessage::Ping(data))) => {
-                        let _ = conn.send_tx.try_send(WsMessage::Pong(data));
+                        // Send Pong through the control channel — priority
+                        // delivery even when the data buffer is full (Bug 7 fix).
+                        if conn.ctrl_tx.try_send(WsMessage::Pong(data)).is_err() {
+                            // Control channel full means the socket writer is
+                            // completely stalled — treat as terminal.
+                            warn!(conn_id = %conn.conn_id, "control channel full — cannot send Pong, closing");
+                            break;
+                        }
                     }
                     Some(Ok(WsMessage::Close(_))) | None => {
                         debug!("WebSocket closed by client");

crates/sprout-relay/src/state.rs

@@ -1,5 +1,6 @@
 //! Shared application state — Arc-wrapped, shared across all connections.
 
+use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
 
@@ -19,13 +20,21 @@ use sprout_workflow::WorkflowEngine;
 
 use crate::api::tokens::MintRateLimiter;
 use crate::config::Config;
+use crate::connection::SLOW_CLIENT_GRACE_LIMIT;
 use crate::subscription::SubscriptionRegistry;
 
+/// Per-connection entry in the connection manager.
+struct ConnEntry {
+    tx: mpsc::Sender<WsMessage>,
+    cancel: CancellationToken,
+    /// Shared with `ConnectionState` — both direct sends and fan-out
+    /// broadcasts track the same consecutive-full counter.
+    backpressure_count: Arc<AtomicU8>,
+}
+
 /// Tracks active WebSocket connections and provides message routing by connection ID.
 pub struct ConnectionManager {
-    /// Map from connection ID to the sender half of the connection's outbound channel
-    /// and the cancellation token for the connection.
-    connections: DashMap<Uuid, (mpsc::Sender<WsMessage>, CancellationToken)>,
+    connections: DashMap<Uuid, ConnEntry>,
 }
 
 impl ConnectionManager {
@@ -36,9 +45,23 @@ impl ConnectionManager {
         }
     }
 
-    /// Registers a connection with its outbound sender and cancellation token.
-    pub fn register(&self, conn_id: Uuid, tx: mpsc::Sender<WsMessage>, cancel: CancellationToken) {
-        self.connections.insert(conn_id, (tx, cancel));
+    /// Registers a connection with its outbound sender, cancellation token,
+    /// and shared backpressure counter (same `Arc` as `ConnectionState`).
+    pub fn register(
+        &self,
+        conn_id: Uuid,
+        tx: mpsc::Sender<WsMessage>,
+        cancel: CancellationToken,
+        backpressure_count: Arc<AtomicU8>,
+    ) {
+        self.connections.insert(
+            conn_id,
+            ConnEntry {
+                tx,
+                cancel,
+                backpressure_count,
+            },
+        );
     }
 
     /// Removes a connection from the registry.
@@ -49,15 +72,24 @@ impl ConnectionManager {
     /// Sends a text message to the given connection.
     ///
     /// Returns `false` if the connection is gone or the buffer is full.
-    /// On a full buffer, cancels the slow client's connection to prevent silent drops.
+    /// On sustained backpressure (>[`SLOW_CLIENT_GRACE_LIMIT`] consecutive full
+    /// buffers), cancels the connection. Transient stalls get a warning only.
     pub fn send_to(&self, conn_id: Uuid, msg: String) -> bool {
         if let Some(entry) = self.connections.get(&conn_id) {
-            let (tx, cancel) = entry.value();
-            match tx.try_send(WsMessage::Text(msg.into())) {
-                Ok(_) => true,
+            let conn = entry.value();
+            match conn.tx.try_send(WsMessage::Text(msg.into())) {
+                Ok(_) => {
+                    conn.backpressure_count.store(0, Ordering::Relaxed);
+                    true
+                }
                 Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
-                    tracing::warn!(conn_id = %conn_id, "fan-out: send buffer full — cancelling slow client");
-                    cancel.cancel();
+                    let count = conn.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1;
+                    if count >= SLOW_CLIENT_GRACE_LIMIT {
+                        tracing::warn!(conn_id = %conn_id, count, "fan-out: sustained backpressure — cancelling slow client");
+                        conn.cancel.cancel();
+                    } else {
+                        tracing::warn!(conn_id = %conn_id, count, grace = SLOW_CLIENT_GRACE_LIMIT, "fan-out: send buffer full — grace {count}/{SLOW_CLIENT_GRACE_LIMIT}");
+                    }
                     false
                 }
                 Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
@@ -175,3 +207,129 @@ impl std::fmt::Debug for AppState {
             .finish()
     }
 }
+
+// ── Unit tests ────────────────────────────────────────────────────────────────
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::connection::{AuthState, ConnectionState};
+    use std::collections::HashMap;
+    use tokio::sync::{Mutex, RwLock};
+
+    /// Helper: create a ConnectionManager with one registered connection.
+    /// Returns (manager, conn_id, receiver, cancel, shared_backpressure_count).
+    fn setup_conn(
+        buffer_size: usize,
+    ) -> (
+        ConnectionManager,
+        Uuid,
+        mpsc::Receiver<WsMessage>,
+        CancellationToken,
+        Arc<AtomicU8>,
+    ) {
+        let mgr = ConnectionManager::new();
+        let conn_id = Uuid::new_v4();
+        let (tx, rx) = mpsc::channel(buffer_size);
+        let cancel = CancellationToken::new();
+        let bp = Arc::new(AtomicU8::new(0));
+        mgr.register(conn_id, tx, cancel.clone(), Arc::clone(&bp));
+        (mgr, conn_id, rx, cancel, bp)
+    }
+
+    #[test]
+    fn send_to_resets_grace_counter_on_success() {
+        let (mgr, id, _rx, _cancel, bp) = setup_conn(16);
+        // Simulate prior backpressure.
+        bp.store(2, Ordering::Relaxed);
+        assert!(mgr.send_to(id, "hello".into()));
+        assert_eq!(
+            bp.load(Ordering::Relaxed),
+            0,
+            "successful send should reset counter"
+        );
+    }
+
+    #[test]
+    fn send_to_increments_grace_counter_on_full() {
+        // Buffer size 1 — fill it, then the next send is Full.
+        let (mgr, id, _rx, cancel, bp) = setup_conn(1);
+        assert!(mgr.send_to(id, "fill".into()));
+        // Buffer is now full.
+        assert!(!mgr.send_to(id, "overflow-1".into()));
+        assert_eq!(bp.load(Ordering::Relaxed), 1, "first overflow → count=1");
+        assert!(
+            !cancel.is_cancelled(),
+            "should not cancel on first overflow"
+        );
+
+        assert!(!mgr.send_to(id, "overflow-2".into()));
+        assert_eq!(bp.load(Ordering::Relaxed), 2);
+        assert!(
+            !cancel.is_cancelled(),
+            "should not cancel on second overflow"
+        );
+    }
+
+    #[test]
+    fn send_to_cancels_after_grace_limit() {
+        let (mgr, id, _rx, cancel, _bp) = setup_conn(1);
+        assert!(mgr.send_to(id, "fill".into()));
+        // Exhaust grace: 3 consecutive Full events.
+        for _ in 0..SLOW_CLIENT_GRACE_LIMIT {
+            mgr.send_to(id, "overflow".into());
+        }
+        assert!(
+            cancel.is_cancelled(),
+            "should cancel after SLOW_CLIENT_GRACE_LIMIT overflows"
+        );
+    }
+
+    #[test]
+    fn shared_counter_between_direct_and_fanout() {
+        // Verify that ConnectionState::send() and ConnectionManager::send_to()
+        // share the same backpressure counter via Arc<AtomicU8>.
+        let conn_id = Uuid::new_v4();
+        let (tx, _rx) = mpsc::channel(1);
+        let (ctrl_tx, _ctrl_rx) = mpsc::channel(8);
+        let cancel = CancellationToken::new();
+        let bp = Arc::new(AtomicU8::new(0));
+
+        let conn = ConnectionState {
+            conn_id,
+            remote_addr: "127.0.0.1:1234".parse().unwrap(),
+            auth_state: RwLock::new(AuthState::Failed),
+            subscriptions: Mutex::new(HashMap::new()),
+            send_tx: tx.clone(),
+            ctrl_tx,
+            cancel: cancel.clone(),
+            backpressure_count: Arc::clone(&bp),
+        };
+
+        let mgr = ConnectionManager::new();
+        mgr.register(conn_id, tx, cancel.clone(), Arc::clone(&bp));
+
+        // Fill the buffer via direct send.
+        assert!(conn.send("fill".into()));
+        // Overflow via fan-out.
+        assert!(!mgr.send_to(conn_id, "overflow-fanout".into()));
+        assert_eq!(
+            bp.load(Ordering::Relaxed),
+            1,
+            "fan-out overflow increments shared counter"
+        );
+        // Overflow via direct send.
+        assert!(!conn.send("overflow-direct".into()));
+        assert_eq!(
+            bp.load(Ordering::Relaxed),
+            2,
+            "direct overflow increments same counter"
+        );
+        // One more fan-out overflow → should cancel (3 consecutive).
+        mgr.send_to(conn_id, "overflow-final".into());
+        assert!(
+            cancel.is_cancelled(),
+            "shared counter reached limit via mixed path"
+        );
+    }
+}

desktop/src/features/messages/ui/MessageComposer.tsx

@@ -136,18 +136,48 @@ export function MessageComposer({
     [content, mentionStartIndex],
   );
 
-  const extractMentionPubkeys = React.useCallback((text: string): string[] => {
-    const pubkeys: string[] = [];
-    const mentionMap = mentionMapRef.current;
+  const extractMentionPubkeys = React.useCallback(
+    (text: string): string[] => {
+      const pubkeys: string[] = [];
 
-    for (const [displayName, pubkey] of mentionMap) {
-      if (text.includes(`@${displayName}`)) {
-        pubkeys.push(pubkey);
+      /** Test whether `@name` appears as a mention token in `text`.
+       *  Left boundary: start-of-string or whitespace.
+       *  Right boundary: end-of-string, whitespace, or punctuation.
+       *  Case-insensitive. Prevents @Ann from matching @Anna or foo@Ann. */
+      const hasMention = (name: string): boolean => {
+        const escaped = name.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+        const pattern = new RegExp(
+          `(?:^|\\s)@${escaped}(?=[\\s,;.!?:)\\]}]|$)`,
+          "i",
+        );
+        return pattern.test(text);
+      };
+
+      // 1. Check the autocomplete-inserted mention map.
+      for (const [displayName, pubkey] of mentionMapRef.current) {
+        if (hasMention(displayName)) {
+          pubkeys.push(pubkey);
+        }
       }
-    }
 
-    return [...new Set(pubkeys)];
-  }, []);
+      // 2. Fall back to scanning channel members for manually-typed @mentions
+      //    that bypassed autocomplete (e.g. user typed "@Daphne" without selecting).
+      for (const member of members) {
+        if (pubkeys.includes(member.pubkey)) {
+          continue; // already matched via autocomplete map
+        }
+        const name =
+          member.displayName ??
+          managedAgentNamesByPubkey.get(member.pubkey.toLowerCase());
+        if (name && hasMention(name)) {
+          pubkeys.push(member.pubkey);
+        }
+      }
+
+      return [...new Set(pubkeys)];
+    },
+    [members, managedAgentNamesByPubkey],
+  );
 
   const submitMessage = React.useCallback(async () => {
     const trimmed = content.trim();