fix(acp): deliver plain-text replies for self-hosted OpenAI-compatible agents

crates/buzz-acp/src/acp.rs

@@ -210,12 +210,62 @@ pub struct AcpClient {
     steer_rx: Option<tokio::sync::mpsc::Receiver<crate::pool::SteerRequest>>,
     /// Usage tracker for goose/buzz-agent's cumulative notification format.
     goose_usage: UsageTracker,
+    /// Plain assistant text streamed during the current turn, and whether the
+    /// agent published a reply itself. Small local models routinely finish a
+    /// multi-step turn by writing the answer as prose instead of calling
+    /// `send_message`; ACP treats streamed text as observability only, so that
+    /// answer is dropped. The pool's mesh-gated delivery fallback consumes
+    /// these via [`take_undelivered_turn_text`](Self::take_undelivered_turn_text).
+    turn_text: String,
+    /// Whether a message-publish tool call was observed during this turn.
+    turn_published: bool,
     /// Per-turn prompt-response usage and Claude's optional cumulative cost.
     standard_usage: StandardUsageTracker,
     /// Known adapter identity for prompt-response usage mapping.
     standard_adapter: Option<StandardAdapterKind>,
 }
 
+/// Maximum assistant text retained per turn for the delivery fallback.
+///
+/// Bounded so a rogue agent streaming forever cannot grow the buffer without
+/// limit; a chat reply that exceeds this is truncated rather than dropped.
+const MAX_TURN_TEXT_BYTES: usize = 8 * 1024;
+
+/// Whether an ACP `tool_call` update represents the agent publishing a message.
+///
+/// Matches both shapes an agent can use: the first-class `send_message` tool
+/// (any `dev__`/server prefix) and a shell command that runs the CLI directly.
+fn is_message_publish(title: &str, raw_input: &str) -> bool {
+    title.ends_with("send_message")
+        || raw_input.contains("messages send")
+        || raw_input.contains("social publish")
+}
+
+/// Whether text is a bare acknowledgement not worth publishing as a reply.
+///
+/// Guards the fallback against posting filler like "OK" or "Done." when the
+/// agent had nothing substantive to say.
+fn is_bare_acknowledgement(text: &str) -> bool {
+    let normalized: String = text
+        .chars()
+        .filter(|c| c.is_alphanumeric() || c.is_whitespace())
+        .collect::<String>()
+        .trim()
+        .to_ascii_lowercase();
+    matches!(
+        normalized.as_str(),
+        "" | "ok"
+            | "okay"
+            | "done"
+            | "sure"
+            | "got it"
+            | "understood"
+            | "acknowledged"
+            | "will do"
+            | "thanks"
+    )
+}
+
 /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape
 /// collisions.  When both sides have an object for the same key, the merge recurses so
 /// unrelated nested keys from `base` are preserved.
@@ -561,6 +611,8 @@ impl AcpClient {
             steering_supported: false,
             steer_rx: None,
             goose_usage: UsageTracker::default(),
+            turn_text: String::new(),
+            turn_published: false,
             standard_usage: StandardUsageTracker::default(),
             standard_adapter,
         })
@@ -791,6 +843,11 @@ impl AcpClient {
         self.goose_usage.begin_turn(session_id);
         self.standard_usage.begin_turn(session_id);
 
+        // Reset per-turn delivery-fallback state alongside usage: text streamed
+        // by a previous turn must never be republished by this one.
+        self.turn_text.clear();
+        self.turn_published = false;
+
         self.last_prompt_id = Some(self.next_id);
         let id = self.next_id;
         self.next_id += 1;
@@ -901,6 +958,23 @@ impl AcpClient {
         self.standard_usage.seed_zero_baseline(session_id);
     }
 
+    /// Take the turn's assistant text if the agent never published a reply.
+    ///
+    /// Returns `None` when the agent published via `send_message` (or the CLI)
+    /// itself, or when the only text was a bare acknowledgement. Consuming
+    /// clears the buffer so the same text can never be posted twice.
+    pub fn take_undelivered_turn_text(&mut self) -> Option<String> {
+        let text = std::mem::take(&mut self.turn_text);
+        if self.turn_published {
+            return None;
+        }
+        let trimmed = text.trim();
+        if trimmed.is_empty() || is_bare_acknowledgement(trimmed) {
+            return None;
+        }
+        Some(trimmed.to_string())
+    }
+
     /// Install a per-turn steer request channel for goose-native
     /// non-cancelling mid-turn delivery.
     ///
@@ -1756,6 +1830,15 @@ impl AcpClient {
             "agent_message_chunk" => {
                 if let Some(text) = update["content"]["text"].as_str() {
                     tracing::info!(target: "acp::stream", "{text}");
+                    // Retain for the mesh-gated delivery fallback (bounded).
+                    let remaining = MAX_TURN_TEXT_BYTES.saturating_sub(self.turn_text.len());
+                    if remaining > 0 {
+                        let mut end = text.len().min(remaining);
+                        while end > 0 && !text.is_char_boundary(end) {
+                            end -= 1;
+                        }
+                        self.turn_text.push_str(&text[..end]);
+                    }
                 }
                 false
             }
@@ -1769,6 +1852,15 @@ impl AcpClient {
                     .and_then(|v| v.as_str())
                     .unwrap_or("unknown");
                 tracing::info!(target: "acp::tool", "tool_call: {title} ({kind})");
+                if !self.turn_published {
+                    let raw_input = update
+                        .get("rawInput")
+                        .map(|v| v.to_string())
+                        .unwrap_or_default();
+                    if is_message_publish(title, &raw_input) {
+                        self.turn_published = true;
+                    }
+                }
                 true
             }
             "tool_call_update" => {
@@ -2330,6 +2422,43 @@ fn configure_no_window(cmd: &mut tokio::process::Command) {
 mod tests {
     use super::*;
 
+    #[test]
+    fn publish_detection_matches_tool_and_shell_shapes() {
+        // First-class tool, with the MCP server prefix the agent reports.
+        assert!(is_message_publish("dev__send_message", "{}"));
+        assert!(is_message_publish("send_message", "{}"));
+        // Shell shapes: the CLI invoked directly.
+        assert!(is_message_publish(
+            "dev__shell",
+            r#"{"command":"buzz messages send --channel c --content hi"}"#
+        ));
+        assert!(is_message_publish(
+            "dev__shell",
+            r#"{"command":"buzz social publish --content hi"}"#
+        ));
+        // Unrelated tools must not suppress the fallback.
+        assert!(!is_message_publish("dev__todo", "{}"));
+        assert!(!is_message_publish("dev__shell", r#"{"command":"ls -la"}"#));
+        // `read_file` ends in neither name and must not match.
+        assert!(!is_message_publish("dev__read_file", r#"{"path":"a"}"#));
+    }
+
+    #[test]
+    fn bare_acknowledgements_are_not_worth_publishing() {
+        for text in ["OK", "ok.", "Done!", "  Sure  ", "Got it", "will do", ""] {
+            assert!(
+                is_bare_acknowledgement(text),
+                "expected {text:?} to be a bare ack"
+            );
+        }
+        for text in ["12", "The capital is Paris.", "OK, the answer is 12"] {
+            assert!(
+                !is_bare_acknowledgement(text),
+                "expected {text:?} to be substantive"
+            );
+        }
+    }
+
     #[test]
     fn stop_reason_parses_all_known_values() {
         assert_eq!(StopReason::from_str("end_turn"), Some(StopReason::EndTurn));

crates/buzz-acp/src/config.rs

@@ -409,6 +409,17 @@ pub struct CliArgs {
     #[arg(long, env = "BUZZ_ACP_NO_BASE_PROMPT")]
     pub no_base_prompt: bool,
 
+    /// Publish the agent's plain reply text when it ends a turn without calling
+    /// `send_message`.
+    ///
+    /// Off by default: capable models always publish their own replies, and
+    /// posting streamed text unconditionally would double-post. Small local
+    /// models served over shared compute reliably finish multi-step turns by
+    /// writing the answer as prose instead, so the desktop shared-compute
+    /// preset opts in.
+    #[arg(long, env = "BUZZ_ACP_DELIVER_PLAIN_REPLIES")]
+    pub deliver_plain_replies: bool,
+
     /// Path to a custom base prompt file. Overrides the compiled-in default.
     /// Mutually exclusive with --no-base-prompt.
     #[arg(
@@ -579,6 +590,9 @@ pub struct Config {
     /// `from_cli()`. `None` when using the compiled-in default or when
     /// `--no-base-prompt` is set.
     pub base_prompt_content: Option<String>,
+    /// Publish the agent's plain reply text when a turn ends without a
+    /// `send_message` call. Opt-in; see the CLI flag for rationale.
+    pub deliver_plain_replies: bool,
 }
 
 /// Maximum length, in characters, of a session title sent to the adapter.
@@ -1122,6 +1136,7 @@ impl Config {
             agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()),
             no_base_prompt: args.no_base_prompt,
             base_prompt_content,
+            deliver_plain_replies: args.deliver_plain_replies,
         };
 
         Ok(config)
@@ -1493,6 +1508,7 @@ mod tests {
             idle_pool_sleep_secs: 0,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }

crates/buzz-acp/src/lib.rs

@@ -2205,6 +2205,7 @@ async fn tokio_main() -> Result<()> {
             .as_deref()
             .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()),
         memory_enabled: config.memory_enabled,
+        deliver_plain_replies: config.deliver_plain_replies,
         harness_name: crate::config::normalize_agent_command_identity(&config.agent_command),
         relay_url: config.relay_url.clone(),
     });
@@ -7152,6 +7153,7 @@ mod build_mcp_servers_tests {
             idle_pool_sleep_secs: 0,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }
@@ -7375,6 +7377,7 @@ mod error_outcome_emission_tests {
             idle_pool_sleep_secs: 0,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }