Add Chats mode

crates/buzz-acp/src/acp.rs

@@ -19,6 +19,9 @@ use crate::observer::{ObserverContext, ObserverHandle};
 /// Lines exceeding this limit are rejected to prevent OOM from rogue agents.
 const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB
 
+/// Byte cap for [`AcpClient::begin_message_capture`] accumulation.
+const MESSAGE_CAPTURE_MAX_LEN: usize = 4_096;
+
 /// An MCP server configuration passed to `session/new`.
 ///
 /// Corresponds to the `McpServerStdio` variant in the ACP schema.
@@ -167,6 +170,11 @@ pub struct AcpClient {
     /// outside of a goose-native turn — the read loop's steer arm is
     /// disabled in that case.
     steer_rx: Option<tokio::sync::mpsc::Receiver<crate::pool::SteerRequest>>,
+    /// Accumulates `agent_message_chunk` text while a capture is active.
+    /// Used by side prompts (e.g. chat-title generation) that need the
+    /// assistant's reply text, which is otherwise only streamed to logs.
+    /// `None` = capture inactive.
+    message_capture: Option<String>,
 }
 
 impl AcpClient {
@@ -258,6 +266,7 @@ impl AcpClient {
             observer_context: ObserverContext::default(),
             active_run_id: None,
             steer_rx: None,
+            message_capture: None,
         })
     }
 
@@ -294,6 +303,20 @@ impl AcpClient {
         }
     }
 
+    /// Start capturing streamed assistant text (`agent_message_chunk`).
+    ///
+    /// Used by side prompts (e.g. chat-title generation) whose reply the
+    /// harness needs as a string. Call before `session_prompt_*`, then
+    /// [`take_message_capture`](Self::take_message_capture) after it returns.
+    pub fn begin_message_capture(&mut self) {
+        self.message_capture = Some(String::new());
+    }
+
+    /// Stop capturing and return the accumulated assistant text.
+    pub fn take_message_capture(&mut self) -> Option<String> {
+        self.message_capture.take()
+    }
+
     /// Send the `initialize` request and return the agent's response result value.
     ///
     /// Must be called exactly once, before any other ACP method.
@@ -1218,6 +1241,18 @@ impl AcpClient {
             "agent_message_chunk" => {
                 if let Some(text) = update["content"]["text"].as_str() {
                     tracing::info!(target: "acp::stream", "{text}");
+                    if let Some(capture) = self.message_capture.as_mut() {
+                        // Side prompts expect short replies; the cap bounds a
+                        // runaway stream, not legitimate output.
+                        if capture.len() < MESSAGE_CAPTURE_MAX_LEN {
+                            let budget = MESSAGE_CAPTURE_MAX_LEN - capture.len();
+                            let mut end = budget.min(text.len());
+                            while end > 0 && !text.is_char_boundary(end) {
+                                end -= 1;
+                            }
+                            capture.push_str(&text[..end]);
+                        }
+                    }
                 }
                 false
             }

crates/buzz-acp/src/base_prompt.md

@@ -47,7 +47,8 @@ All replies and delegations — including task assignments to other agents — g
 ### General
 
 - Respond promptly to @mentions. Be direct — no preamble. Name what you did, what you found, or what you need.
-- **Every turn that processes a user message MUST end with `buzz messages send`.** Your reasoning and tool calls are invisible to users — if you didn't send a message, they saw nothing. A turn that ends without a sent message is a silent failure.
+- **Every turn that processes a user message MUST end with `buzz messages send --channel <UUID> --content "..."`.** Message text is not positional — always pass it with `--content`, or pipe it with `--content -` for shell-sensitive text. Your reasoning and tool calls are invisible to users — if you didn't send a message, they saw nothing. A turn that ends without a sent message is a silent failure.
+- Do not narrate that you are about to send a reply, or that you sent one. Put the actual human-facing answer in the final `buzz messages send` content only.
 - For work that requires follow-up tools, create an open todo **before** sending the pickup acknowledgment. Keep it open until the deliverable is verified and you have sent a completion or blocker message; never end a turn with open todo state unless you have posted that completion or blocker message.
 - Use GitHub-flavored Markdown. Fenced code blocks with language tags for syntax highlighting.
 - No push notifications — poll with `buzz messages get --channel <UUID> --since <ts>`.

crates/buzz-acp/src/config.rs

@@ -436,6 +436,16 @@ pub struct ChannelFilter {
     pub require_mention: bool,
 }
 
+/// Chat channels are single-agent addressed even when multiple agents are
+/// members: the desktop client writes the default agent as an implicit `p` tag
+/// and explicit @mentions add more `p` tags. Force `#p` filtering so broad
+/// agent configs do not fan every chat turn out to every attached agent.
+pub fn force_mention_filter_for_chat(filter: &mut ChannelFilter, channel_type: Option<&str>) {
+    if channel_type == Some("chat") {
+        filter.require_mention = true;
+    }
+}
+
 #[derive(Debug)]
 pub struct Config {
     pub keys: Keys,
@@ -1414,6 +1424,30 @@ mod tests {
         assert!(!f.require_mention);
     }
 
+    #[test]
+    fn test_chat_channels_force_mention_filter() {
+        let mut filter = ChannelFilter {
+            kinds: None,
+            require_mention: false,
+        };
+
+        force_mention_filter_for_chat(&mut filter, Some("chat"));
+
+        assert!(filter.require_mention);
+    }
+
+    #[test]
+    fn test_non_chat_channels_keep_resolved_filter() {
+        let mut filter = ChannelFilter {
+            kinds: None,
+            require_mention: false,
+        };
+
+        force_mention_filter_for_chat(&mut filter, Some("stream"));
+
+        assert!(!filter.require_mention);
+    }
+
     #[test]
     fn normalizes_goose_args_to_acp() {
         assert_eq!(normalize_agent_args("goose", Vec::new()), vec!["acp"]);

crates/buzz-acp/src/lib.rs

@@ -77,6 +77,129 @@ async fn publish_presence(
     Ok(())
 }
 
+fn channel_is_chat(channel_info: &HashMap<Uuid, relay::ChannelInfo>, channel_id: Uuid) -> bool {
+    channel_info
+        .get(&channel_id)
+        .is_some_and(|info| info.channel_type == "chat")
+}
+
+/// Oldest chat backlog the startup replay will deliver. Bounds necro-replies
+/// to abandoned chats while comfortably covering the activation flow (owner
+/// messages a stopped agent, then starts it from the desktop).
+const CHAT_BACKLOG_MAX_AGE_SECS: u64 = 6 * 60 * 60;
+
+/// Replay floor for a chat channel's startup subscription.
+///
+/// A chat's owner often messages while the agent is stopped — the desktop's
+/// "Activate agent" card starts the agent on demand — and a `since: now`
+/// subscription would silently skip that pending message forever. Replay from
+/// just after the agent's own last reply in the channel (it has answered
+/// everything before that), capped to a recent window. Best-effort: on query
+/// failure, fall back to plain live-only subscription.
+async fn chat_backlog_replay_since(
+    rest: &relay::RestClient,
+    channel_id: Uuid,
+    agent_pubkey_hex: &str,
+    startup_watermark: u64,
+) -> Option<u64> {
+    let floor = startup_watermark.saturating_sub(CHAT_BACKLOG_MAX_AGE_SECS);
+    let author = nostr::PublicKey::from_hex(agent_pubkey_hex).ok()?;
+    let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
+    let filter = nostr::Filter::new()
+        .kind(nostr::Kind::Custom(9))
+        .author(author)
+        .custom_tags(h_tag, [channel_id.to_string()])
+        .limit(1);
+
+    match rest.query(&[filter]).await {
+        Ok(value) => {
+            let last_reply_ts = value.as_array().and_then(|events| {
+                events
+                    .iter()
+                    .filter_map(|event| event.get("created_at").and_then(|v| v.as_u64()))
+                    .max()
+            });
+            Some(match last_reply_ts {
+                Some(ts) => ts.saturating_add(1).max(floor),
+                None => floor,
+            })
+        }
+        Err(error) => {
+            tracing::debug!("chat backlog query failed for {channel_id}: {error}");
+            None
+        }
+    }
+}
+
+/// Reclassify compatibility chats.
+///
+/// Relays that predate `channel_type: chat` store chats as plain private
+/// channels — the chat-ness lives only in a kind 30623 (or legacy 30078)
+/// chat-metadata event. Without this, every chat behavior keyed on
+/// `channel_type == "chat"` (activation backlog replay, mention gating,
+/// reaction suppression, title emission) silently degrades against staging.
+async fn mark_compat_chats(
+    rest: &relay::RestClient,
+    channel_info: &mut HashMap<Uuid, relay::ChannelInfo>,
+) {
+    const KIND_CHAT_METADATA: u16 = 30623;
+    const LEGACY_CHAT_METADATA_KIND: u16 = 30078;
+    const LEGACY_D_PREFIX: &str = "buzz:chat:";
+
+    if channel_info
+        .values()
+        .all(|info| info.channel_type == "chat")
+    {
+        return;
+    }
+    let filters = [
+        nostr::Filter::new()
+            .kind(nostr::Kind::Custom(KIND_CHAT_METADATA))
+            .limit(500),
+        nostr::Filter::new()
+            .kind(nostr::Kind::Custom(LEGACY_CHAT_METADATA_KIND))
+            .limit(1000),
+    ];
+    let value = match rest.query(&filters).await {
+        Ok(value) => value,
+        Err(error) => {
+            tracing::debug!("compat chat metadata query failed: {error}");
+            return;
+        }
+    };
+    let mut reclassified = 0usize;
+    for event in value.as_array().into_iter().flatten() {
+        let kind = event.get("kind").and_then(|v| v.as_u64()).unwrap_or(0);
+        let tag_value = |name: &str| -> Option<String> {
+            event.get("tags")?.as_array()?.iter().find_map(|tag| {
+                let tag = tag.as_array()?;
+                (tag.first()?.as_str()? == name)
+                    .then(|| tag.get(1)?.as_str().map(str::to_string))
+                    .flatten()
+            })
+        };
+        let channel_id = if kind == u64::from(LEGACY_CHAT_METADATA_KIND) {
+            tag_value("chat_h").or_else(|| {
+                tag_value("d").and_then(|d| d.strip_prefix(LEGACY_D_PREFIX).map(str::to_string))
+            })
+        } else {
+            tag_value("d")
+        };
+        let Some(channel_id) = channel_id.and_then(|id| id.parse::<Uuid>().ok()) else {
+            continue;
+        };
+        if let Some(info) = channel_info.get_mut(&channel_id) {
+            if info.channel_type != "chat" {
+                info.channel_type = "chat".to_string();
+                reclassified += 1;
+            }
+        }
+    }
+    if reclassified > 0 {
+        tracing::info!("reclassified {reclassified} compatibility chat channel(s)");
+    }
+}
+
 /// Resolve the agent's owner pubkey at startup.
 ///
 /// Priority:
@@ -1311,10 +1434,12 @@ async fn tokio_main() -> Result<()> {
         }
     }
 
-    let channel_info_map = relay
+    let mut channel_info_map = relay
         .discover_channels()
         .await
         .map_err(|e| anyhow::anyhow!("channel discovery error: {e}"))?;
+    mark_compat_chats(&relay.rest_client(), &mut channel_info_map).await;
+    let channel_info_map = channel_info_map;
 
     tracing::info!("discovered {} channel(s)", channel_info_map.len());
     let channel_ids: Vec<Uuid> = channel_info_map.keys().copied().collect();
@@ -1356,13 +1481,36 @@ async fn tokio_main() -> Result<()> {
         }
     };
 
-    let channel_filters = config::resolve_channel_filters(&config, &channel_ids, &rules);
+    let mut channel_filters = config::resolve_channel_filters(&config, &channel_ids, &rules);
+    for (channel_id, filter) in channel_filters.iter_mut() {
+        config::force_mention_filter_for_chat(
+            filter,
+            channel_info_map
+                .get(channel_id)
+                .map(|info| info.channel_type.as_str()),
+        );
+    }
+    let mut event_channel_info = channel_info_map.clone();
     if channel_filters.is_empty() {
         tracing::warn!("no channel subscriptions resolved — agent will sit idle");
     }
+    let backlog_rest = relay.rest_client();
     for (channel_id, filter) in &channel_filters {
-        if let Err(e) = relay.subscribe_channel(*channel_id, filter.clone()).await {
+        // Chats replay their unanswered backlog; other channel types stay
+        // live-only from the startup watermark.
+        let replay_since = if channel_is_chat(&channel_info_map, *channel_id) {
+            chat_backlog_replay_since(&backlog_rest, *channel_id, &pubkey_hex, startup_watermark)
+                .await
+        } else {
+            None
+        };
+        if let Err(e) = relay
+            .subscribe_channel_from(*channel_id, filter.clone(), replay_since)
+            .await
+        {
             tracing::warn!("failed to subscribe to channel {channel_id}: {e}");
+        } else if let Some(since) = replay_since {
+            tracing::info!("subscribed to channel {channel_id} (chat backlog since {since})");
         } else {
             tracing::info!("subscribed to channel {channel_id}");
         }
@@ -1402,6 +1550,7 @@ async fn tokio_main() -> Result<()> {
             .as_deref()
             .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()),
         memory_enabled: config.memory_enabled,
+        titled_channels: std::sync::Mutex::new(std::collections::HashSet::new()),
     });
 
     if !config.memory_enabled {
@@ -1730,7 +1879,29 @@ async fn tokio_main() -> Result<()> {
                                     // stripped for a legitimately re-added channel.
                                     removed_channels.remove(&ch);
 
-                                    if let Some(filter) = config::resolve_dynamic_channel_filter(&config, ch, &rules) {
+                                    if let Some(mut filter) = config::resolve_dynamic_channel_filter(&config, ch, &rules) {
+                                        match relay.discover_channels().await {
+                                            Ok(discovered) => {
+                                                event_channel_info.extend(discovered);
+                                                mark_compat_chats(
+                                                    &ctx.rest_client,
+                                                    &mut event_channel_info,
+                                                )
+                                                .await;
+                                            }
+                                            Err(e) => {
+                                                tracing::debug!(
+                                                    channel_id = %ch,
+                                                    "membership notification: channel metadata refresh failed: {e}"
+                                                );
+                                            }
+                                        }
+                                        config::force_mention_filter_for_chat(
+                                            &mut filter,
+                                            event_channel_info
+                                                .get(&ch)
+                                                .map(|info| info.channel_type.as_str()),
+                                        );
                                         tracing::info!(channel_id = %ch, "membership notification: subscribing to new channel");
                                         if let Err(e) = relay.subscribe_channel_from(ch, filter, Some(ts)).await {
                                             tracing::warn!("failed to subscribe to new channel {ch}: {e}");
@@ -1752,6 +1923,7 @@ async fn tokio_main() -> Result<()> {
                                     // Track removed channels so checked-out agents get
                                     // their sessions stripped when they return to the pool.
                                     removed_channels.insert(ch);
+                                    event_channel_info.remove(&ch);
                                     typing_channels.remove(&ch);
                                     // Best-effort: clean up 👀 on drained events.
                                     // Note: the relay revokes membership before
@@ -1920,6 +2092,17 @@ async fn tokio_main() -> Result<()> {
                                 }
                             }
 
+                            if channel_is_chat(&event_channel_info, buzz_event.channel_id)
+                                && !event_mentions_agent(&buzz_event.event, &pubkey_hex)
+                            {
+                                tracing::debug!(
+                                    channel_id = %buzz_event.channel_id,
+                                    kind = buzz_event.event.kind.as_u16(),
+                                    "chat event did not mention this agent — dropping"
+                                );
+                                continue;
+                            }
+
                             let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex).await;
                             let prompt_tag = match matched {
                                 Some(m) => m.prompt_tag,
@@ -1955,7 +2138,15 @@ async fn tokio_main() -> Result<()> {
                             // Fire-and-forget: on rare fast-failure paths the
                             // guard's cleanup may race with this add, leaving a
                             // cosmetic stale 👀. Acceptable — see ReactionGuard docs.
-                            if accepted {
+                            // Chats skip it: their UI has live working
+                            // indicators, and reaction emoji on the user's
+                            // message reads as noise there.
+                            if accepted
+                                && !channel_is_chat(
+                                    &event_channel_info,
+                                    buzz_event.channel_id,
+                                )
+                            {
                                 let rc = ctx.rest_client.clone();
                                 let eid = event_id_hex.clone();
                                 tokio::spawn(async move {