fix(acp): reconcile channel memberships

crates/buzz-acp/src/lib.rs

@@ -66,6 +66,64 @@ const MODELS_TIMEOUT: Duration = Duration::from_secs(10);
 /// human interaction, so it must not share the short probe timeout.
 const AUTHENTICATE_TIMEOUT: Duration = Duration::from_secs(10 * 60);
 
+/// Return memberships discovered by the REST bridge that are not yet covered
+/// by a live relay subscription.
+///
+/// Reconciliation is intentionally additive. A temporarily stale or partial
+/// REST response must never revoke an existing channel subscription; removals
+/// continue to be driven by signed membership notifications.
+fn channel_memberships_to_reconcile(
+    discovered: &HashMap<Uuid, relay::ChannelInfo>,
+    subscribed: &HashSet<Uuid>,
+) -> Vec<Uuid> {
+    let mut missing: Vec<_> = discovered
+        .keys()
+        .filter(|channel_id| !subscribed.contains(channel_id))
+        .copied()
+        .collect();
+    missing.sort_unstable();
+    missing
+}
+
+#[cfg(test)]
+mod membership_reconciliation_tests {
+    use super::*;
+
+    fn channel(name: &str) -> relay::ChannelInfo {
+        relay::ChannelInfo {
+            name: name.to_string(),
+            channel_type: "dm".to_string(),
+        }
+    }
+
+    #[test]
+    fn returns_only_unsubscribed_memberships_in_stable_order() {
+        let first = Uuid::from_u128(1);
+        let second = Uuid::from_u128(2);
+        let third = Uuid::from_u128(3);
+        let discovered = HashMap::from([
+            (third, channel("third")),
+            (first, channel("first")),
+            (second, channel("second")),
+        ]);
+        let subscribed = HashSet::from([second]);
+
+        assert_eq!(
+            channel_memberships_to_reconcile(&discovered, &subscribed),
+            vec![first, third]
+        );
+    }
+
+    #[test]
+    fn an_incomplete_snapshot_never_requests_unsubscription() {
+        let subscribed_only = Uuid::from_u128(1);
+        let subscribed = HashSet::from([subscribed_only]);
+
+        assert!(channel_memberships_to_reconcile(&HashMap::new(), &subscribed).is_empty());
+        assert!(subscribed.contains(&subscribed_only));
+    }
+}
+
 /// Publish a kind:20001 presence update event via the WebSocket connection.
 ///
 /// Ephemeral kinds (20000-29999) are rejected by the HTTP bridge, so presence
@@ -1978,6 +2036,22 @@ async fn tokio_main() -> Result<()> {
     // and capture it in TaskMeta at dispatch time.
     let mut removed_channels: HashSet<Uuid> = HashSet::new();
 
+    // Membership notifications are the primary source of truth, but a client
+    // can miss one while reconnecting. Periodically compare the authenticated
+    // REST membership view with active subscriptions so newly created DMs and
+    // groups become usable without restarting the harness.
+    let reconcile_period_secs = std::env::var("BUZZ_ACP_CHANNEL_RECONCILE_SECS")
+        .ok()
+        .and_then(|value| value.parse::<u64>().ok())
+        .unwrap_or(30)
+        .max(5);
+    let mut channel_reconcile = tokio::time::interval_at(
+        tokio::time::Instant::now() + Duration::from_secs(reconcile_period_secs),
+        Duration::from_secs(reconcile_period_secs),
+    );
+    channel_reconcile.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let membership_rest = relay.rest_client();
+
     //
     // One SlotCircuit per agent slot. crash_times entries are pruned to the last
     // CIRCUIT_BREAKER_WINDOW on each respawn attempt. The Vec is indexed by
@@ -2000,6 +2074,7 @@ async fn tokio_main() -> Result<()> {
         Panic(tokio::task::JoinError),
         SteerAck(SteerAckEvent),
         Wake(u32, Result<AgentPool, String>),
+        MembershipSnapshot(Result<HashMap<Uuid, relay::ChannelInfo>, relay::RelayError>),
     }
 
     loop {
@@ -2203,6 +2278,12 @@ async fn tokio_main() -> Result<()> {
                     }
                     None
                 }
+                _ = channel_reconcile.tick() => {
+                    let _ = result_rx;
+                    Some(PoolEvent::MembershipSnapshot(
+                        membership_rest.discover_channels().await,
+                    ))
+                }
                 // Remaining branches don't touch pool — evaluated when pool is idle.
                 buzz_event = relay.next_event() => {
                     let _ = result_rx; // end split borrow before relay handling
@@ -2925,6 +3006,40 @@ async fn tokio_main() -> Result<()> {
                     }
                 }
             }
+            Some(PoolEvent::MembershipSnapshot(Ok(discovered))) => {
+                for channel_id in
+                    channel_memberships_to_reconcile(&discovered, &subscribed_channel_ids)
+                {
+                    let Some(filter) =
+                        config::resolve_dynamic_channel_filter(&config, channel_id, &rules)
+                    else {
+                        tracing::debug!(
+                            %channel_id,
+                            "membership reconciliation: no matching rules — skipping"
+                        );
+                        continue;
+                    };
+
+                    tracing::info!(
+                        %channel_id,
+                        "membership reconciliation: subscribing to discovered channel"
+                    );
+                    match relay.subscribe_channel(channel_id, filter).await {
+                        Ok(()) => {
+                            subscribed_channel_ids.insert(channel_id);
+                            removed_channels.remove(&channel_id);
+                        }
+                        Err(error) => tracing::warn!(
+                            %channel_id,
+                            %error,
+                            "membership reconciliation: subscription failed"
+                        ),
+                    }
+                }
+            }
+            Some(PoolEvent::MembershipSnapshot(Err(error))) => {
+                tracing::warn!(%error, "membership reconciliation failed");
+            }
             None => {} // relay/heartbeat/shutdown branches handled inline above
         }
     }

crates/buzz-acp/src/relay.rs

@@ -241,6 +241,58 @@ pub struct RestClient {
     pub auth_tag_json: Option<String>,
 }
 
+async fn discover_channels_with_rest(
+    rest: &RestClient,
+) -> Result<HashMap<Uuid, ChannelInfo>, RelayError> {
+    use nostr::{Alphabet, SingleLetterTag};
+
+    let pk_hex = rest.keys.public_key().to_hex();
+    let p_tag = SingleLetterTag::lowercase(Alphabet::P);
+    let member_filter = nostr::Filter::new()
+        .kind(Kind::Custom(
+            buzz_core::kind::KIND_NIP29_GROUP_MEMBERS as u16,
+        ))
+        .custom_tags(p_tag, [pk_hex.as_str()]);
+    let member_events = rest.query(&[member_filter]).await?;
+    let member_arr = member_events
+        .as_array()
+        .ok_or_else(|| RelayError::Http("expected JSON array from /query (members)".into()))?;
+
+    let mut channel_uuids = Vec::new();
+    for event in member_arr {
+        if let Some(tags) = event.get("tags").and_then(|tags| tags.as_array()) {
+            for tag in tags {
+                if let Some(parts) = tag.as_array() {
+                    if parts.first().and_then(|value| value.as_str()) == Some("d") {
+                        if let Some(value) = parts.get(1).and_then(|value| value.as_str()) {
+                            if let Ok(channel_id) = value.parse::<Uuid>() {
+                                channel_uuids.push(channel_id);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    if channel_uuids.is_empty() {
+        debug!("discovered 0 channel(s)");
+        return Ok(HashMap::new());
+    }
+
+    let d_tag = SingleLetterTag::lowercase(Alphabet::D);
+    let d_values: Vec<String> = channel_uuids.iter().map(ToString::to_string).collect();
+    let meta_filter = nostr::Filter::new()
+        .kind(Kind::Custom(
+            buzz_core::kind::KIND_NIP29_GROUP_METADATA as u16,
+        ))
+        .custom_tags(d_tag, d_values);
+    let meta_events = rest.query(&[meta_filter]).await?;
+    let map = merge_discovered_channels(channel_uuids, &meta_events);
+    debug!("discovered {} channel(s)", map.len());
+    Ok(map)
+}
+
 /// Whether an HTTP status code is retriable (transient server/rate-limit errors).
 fn is_retriable_status(status: reqwest::StatusCode) -> bool {
     matches!(status.as_u16(), 429 | 502 | 503 | 504)
@@ -262,6 +314,11 @@ fn unix_now_secs() -> u64 {
 }
 
 impl RestClient {
+    /// Discover channels this identity is currently a member of.
+    pub async fn discover_channels(&self) -> Result<HashMap<Uuid, ChannelInfo>, RelayError> {
+        discover_channels_with_rest(self).await
+    }
+
     /// Sign a NIP-98 HTTP Auth event (kind:27235) for the given method/URL/body.
     ///
     /// Returns the `Authorization: Nostr <base64>` header value (without the
@@ -670,62 +727,7 @@ impl HarnessRelay {
     /// the agent pubkey to find channel memberships, then queries kind:39000
     /// (group metadata) for channel names and types.
     pub async fn discover_channels(&self) -> Result<HashMap<Uuid, ChannelInfo>, RelayError> {
-        use nostr::{Alphabet, SingleLetterTag};
-
-        let rest = self.rest_client();
-        let pk_hex = self.keys.public_key().to_hex();
-
-        // Step 1: Find all channels where agent is a member (kind:39002 with #p tag).
-        let p_tag = SingleLetterTag::lowercase(Alphabet::P);
-        let member_filter = nostr::Filter::new()
-            .kind(Kind::Custom(
-                buzz_core::kind::KIND_NIP29_GROUP_MEMBERS as u16,
-            ))
-            .custom_tags(p_tag, [pk_hex.as_str()]);
-        let member_events = rest.query(&[member_filter]).await?;
-
-        let member_arr = member_events
-            .as_array()
-            .ok_or_else(|| RelayError::Http("expected JSON array from /query (members)".into()))?;
-
-        // Extract channel UUIDs from #d tags.
-        let mut channel_uuids: Vec<Uuid> = Vec::new();
-        for ev in member_arr {
-            if let Some(tags) = ev.get("tags").and_then(|t| t.as_array()) {
-                for tag in tags {
-                    if let Some(arr) = tag.as_array() {
-                        if arr.first().and_then(|v| v.as_str()) == Some("d") {
-                            if let Some(d_val) = arr.get(1).and_then(|v| v.as_str()) {
-                                if let Ok(uuid) = d_val.parse::<Uuid>() {
-                                    channel_uuids.push(uuid);
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        }
-
-        if channel_uuids.is_empty() {
-            debug!("discovered 0 channel(s)");
-            return Ok(HashMap::new());
-        }
-
-        // Step 2: Fetch metadata (kind:39000) for discovered channels.
-        let d_tag = SingleLetterTag::lowercase(Alphabet::D);
-        let d_values: Vec<String> = channel_uuids.iter().map(|u| u.to_string()).collect();
-        let meta_filter = nostr::Filter::new()
-            .kind(Kind::Custom(
-                buzz_core::kind::KIND_NIP29_GROUP_METADATA as u16,
-            ))
-            .custom_tags(d_tag, d_values);
-        let meta_events = rest.query(&[meta_filter]).await?;
-
-        // Step 3: Build the final subscribe set, skipping archived channels.
-        let map = merge_discovered_channels(channel_uuids, &meta_events);
-
-        debug!("discovered {} channel(s)", map.len());
-        Ok(map)
+        self.rest_client().discover_channels().await
     }
 
     /// Build a [`RestClient`] that shares this relay's HTTP credentials.