fix(acp): deliver plain replies for shared compute agents

buzz-agent's contract is that output is tool calls: streamed assistant text
is observability only and never published. Capable models honour that and
call `send_message`. Small local models served over shared compute do not
always — on multi-step turns they run their tools, then write the answer as
prose and end the turn, so the turn reports success with nothing in the
channel.

Retain the turn's streamed text in AcpClient (bounded at 8 KiB) alongside a
flag for whether a publish tool call was seen, and on EndTurn publish that
text as a threaded channel reply when the agent published nothing itself.
Gated on EndTurn only: MaxTokens / MaxTurnRequests mean the turn was
truncated, so the text is not a deliberate answer. Bare acknowledgements
("OK", "Done") are filtered, and taking the buffer clears it so the same
text can never post twice.

Off by default via --deliver-plain-replies / BUZZ_ACP_DELIVER_PLAIN_REPLIES:
for cloud models, which reliably publish their own replies, posting streamed
text would double-post. The desktop shared-compute preset opts in, so this
stays scoped to the case that needs it.

Verified end to end against a live relay with stub ACP agents that pin each
branch deterministically: a prose-only agent gets its answer delivered, and
an agent that calls send_message while also streaming prose posts exactly
once with the fallback silent.

Signed-off-by: Michael Neale <michael.neale@gmail.com>

crates/buzz-acp/src/acp.rs

@@ -211,6 +211,56 @@ pub struct AcpClient {
     /// deltas. Both goose and buzz-agent emit this notification; goose gates
     /// on client capability advertisement, buzz-agent emits unconditionally.
     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,
+}
+
+/// 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
@@ -550,6 +600,8 @@ impl AcpClient {
             steering_supported: false,
             steer_rx: None,
             goose_usage: UsageTracker::default(),
+            turn_text: String::new(),
+            turn_published: false,
         })
     }
 
@@ -760,6 +812,11 @@ impl AcpClient {
         // misattributed to this turn.
         self.goose_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;
@@ -864,6 +921,23 @@ impl AcpClient {
         self.goose_usage.take()
     }
 
+    /// 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.
     ///
@@ -1715,6 +1789,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
             }
@@ -1728,6 +1811,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" => {
@@ -2223,6 +2315,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(
@@ -561,6 +572,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.
@@ -1102,6 +1116,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)
@@ -1471,6 +1486,7 @@ mod tests {
             lazy_pool: false,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }

crates/buzz-acp/src/lib.rs

@@ -1557,6 +1557,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(),
     });
@@ -5018,6 +5019,7 @@ mod build_mcp_servers_tests {
             lazy_pool: false,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }
@@ -5239,6 +5241,7 @@ mod error_outcome_emission_tests {
             lazy_pool: false,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }

crates/buzz-acp/src/pool.rs

@@ -543,6 +543,11 @@ pub struct PromptContext {
     /// `[Agent Memory — core]` section. On by default; disabled via
     /// `--no-memory` / `BUZZ_ACP_NO_MEMORY`.
     pub memory_enabled: bool,
+    /// Whether to publish the agent's plain reply text when a turn ends without
+    /// a `send_message` call. Off by default; the desktop shared-compute preset
+    /// opts in because small local models routinely finish a multi-step turn by
+    /// writing the answer as prose instead of calling the tool.
+    pub deliver_plain_replies: bool,
     /// Harness identity string for NIP-AM `harness` field. Derived from the
     /// configured `agent_command` at startup (e.g. `"goose"`, `"buzz-agent"`).
     pub harness_name: String,
@@ -1362,6 +1367,14 @@ pub async fn run_prompt_task(
         .as_ref()
         .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect())
         .unwrap_or_default();
+    // Capture the reply target now: `batch` is moved into the requeue/outcome
+    // arms below, but the plain-reply fallback runs after the turn completes.
+    // Heartbeats have no batch and therefore no fallback target.
+    let plain_reply_target = if ctx.deliver_plain_replies {
+        batch.as_ref().and_then(plain_reply_target_for)
+    } else {
+        None
+    };
     agent.acp.observe(
         "turn_started",
         serde_json::json!({
@@ -2112,6 +2125,31 @@ pub async fn run_prompt_task(
                 agent.state.invalidate(&source);
             }
 
+            // Deliver the turn's plain text when the agent ended the turn
+            // without publishing a reply itself. Gated on EndTurn only:
+            // MaxTokens / MaxTurnRequests mean the turn was truncated, so the
+            // text is not a deliberate answer. `take_*` clears the buffer, so
+            // this can never post the same text twice.
+            if matches!(stop_reason, StopReason::EndTurn) {
+                if let Some(target) = plain_reply_target.as_ref() {
+                    if let Some(text) = agent.acp.take_undelivered_turn_text() {
+                        tracing::warn!(
+                            target: "pool::prompt",
+                            channel = %target.channel_id,
+                            "agent ended turn without publishing; delivering its plain text as a channel reply"
+                        );
+                        post_threaded_message(
+                            &ctx.rest_client,
+                            target.channel_id,
+                            &target.thread_tags,
+                            &text,
+                            "plain-reply delivery",
+                        )
+                        .await;
+                    }
+                }
+            }
+
             let core_stop = acp_stop_to_core(&stop_reason);
             let usage = agent.acp.take_turn_usage();
             publish_agent_turn_metric(
@@ -3584,6 +3622,34 @@ pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str
     }
 }
 
+/// Where a plain-reply fallback should post, captured before `batch` is moved.
+#[derive(Debug, Clone)]
+pub(crate) struct PlainReplyTarget {
+    channel_id: Uuid,
+    thread_tags: ThreadTags,
+}
+
+/// Derive the plain-reply target from the batch's triggering event.
+///
+/// Anchors to the trigger's own thread when it is itself a reply; otherwise the
+/// trigger becomes the thread root, so the fallback reply lands in the same
+/// place the agent's own `send_message` would have.
+pub(crate) fn plain_reply_target_for(batch: &FlushBatch) -> Option<PlainReplyTarget> {
+    let last = batch.events.last()?;
+    let parsed = crate::queue::parse_thread_tags(&last.event);
+    let trigger_hex = last.event.id.to_hex();
+    let root = parsed.root_event_id.clone().unwrap_or(trigger_hex.clone());
+    let parent = parsed.parent_event_id.clone().unwrap_or(trigger_hex);
+    Some(PlainReplyTarget {
+        channel_id: batch.channel_id,
+        thread_tags: ThreadTags {
+            root_event_id: Some(root),
+            parent_event_id: Some(parent),
+            mentioned_pubkeys: Vec::new(),
+        },
+    })
+}
+
 /// Best-effort: post a visible failure notice (kind:9) to a channel after a
 /// batch is dead-lettered. Replies into the thread of `thread_tags` when the
 /// triggering event was threaded. Errors are logged and swallowed — the
@@ -3593,6 +3659,21 @@ pub(crate) async fn post_failure_notice(
     channel_id: Uuid,
     thread_tags: &ThreadTags,
     content: &str,
+) {
+    post_threaded_message(rest, channel_id, thread_tags, content, "failure notice").await;
+}
+
+/// Best-effort: post a threaded kind:9 message to a channel.
+///
+/// Shared by the dead-letter failure notice and the plain-reply delivery
+/// fallback; `what` only labels the log lines so the two are distinguishable.
+/// Errors are logged and swallowed — publishing must never take down the loop.
+async fn post_threaded_message(
+    rest: &crate::relay::RestClient,
+    channel_id: Uuid,
+    thread_tags: &ThreadTags,
+    content: &str,
+    what: &str,
 ) {
     let thread_ref = thread_tags.root_event_id.as_deref().and_then(|root| {
         let root_id = nostr::EventId::from_hex(root).ok()?;
@@ -3610,21 +3691,21 @@ pub(crate) async fn post_failure_notice(
         match buzz_sdk::build_message(channel_id, content, thread_ref.as_ref(), &[], false, &[]) {
             Ok(b) => b,
             Err(e) => {
-                tracing::warn!(channel = %channel_id, "failure notice: build failed: {e}");
+                tracing::warn!(channel = %channel_id, "{what}: build failed: {e}");
                 return;
             }
         };
     let event = match builder.sign_with_keys(&rest.keys) {
         Ok(e) => e,
         Err(e) => {
-            tracing::warn!(channel = %channel_id, "failure notice: sign failed: {e}");
+            tracing::warn!(channel = %channel_id, "{what}: sign failed: {e}");
             return;
         }
     };
     match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await {
         Ok(Ok(_)) => {}
-        Ok(Err(e)) => tracing::warn!(channel = %channel_id, "failure notice failed: {e}"),
-        Err(_) => tracing::warn!(channel = %channel_id, "failure notice timed out"),
+        Ok(Err(e)) => tracing::warn!(channel = %channel_id, "{what} failed: {e}"),
+        Err(_) => tracing::warn!(channel = %channel_id, "{what} timed out"),
     }
 }
 
@@ -5405,6 +5486,7 @@ mod tests {
             agent_keys: agent_keys.clone(),
             agent_owner_pubkey: owner_pubkey,
             memory_enabled: false,
+            deliver_plain_replies: false,
             harness_name: "goose".to_string(),
             relay_url: "ws://127.0.0.1:3000".to_string(),
         }

desktop/src-tauri/src/managed_agents/relay_mesh.rs

@@ -42,6 +42,16 @@ pub fn apply_relay_mesh_env(
         RELAY_MESH_PREFER_MESH_FOR_AUTO_ENV.to_string(),
         "1".to_string(),
     );
+    // Shared compute runs small local models, which reliably finish a
+    // multi-step turn by writing the answer as prose instead of calling
+    // `send_message` — the reply is then never published. Opt this preset into
+    // the harness fallback that delivers that text. Mesh-only: cloud models
+    // publish their own replies, where the fallback would double-post.
+    // Value is `true`, not `1`: these are clap bool flags, which reject `1`.
+    env.insert(
+        "BUZZ_ACP_DELIVER_PLAIN_REPLIES".to_string(),
+        "true".to_string(),
+    );
     // Keep the requested response inside smaller local-model context windows.
     // These are defaults, not policy: the effective agent/persona/global env
     // may deliberately choose a smaller cap or a different effort. This function
@@ -118,6 +128,24 @@ mod tests {
                 .map(String::as_str),
             Some("1")
         );
+        // Small local models end multi-step turns with prose instead of a
+        // `send_message` call, so shared compute opts into harness delivery.
+        assert_eq!(
+            env.get("BUZZ_ACP_DELIVER_PLAIN_REPLIES").map(String::as_str),
+            // `true`, not `1`: clap bool flags reject `1` and the harness
+            // exits at startup with "invalid value '1'".
+            Some("true")
+        );
+    }
+
+    #[test]
+    fn non_mesh_provider_does_not_opt_into_plain_reply_delivery() {
+        let mut env = BTreeMap::new();
+        apply_relay_mesh_env(&mut env, Some("anthropic"), Some("claude-sonnet-4"));
+
+        // Cloud models publish their own replies; delivering streamed text
+        // there would double-post.
+        assert_eq!(env.get("BUZZ_ACP_DELIVER_PLAIN_REPLIES"), None);
     }
 
     #[test]