fix(acp): deliver plain replies for openai-compat/Ollama agents (rebased)

crates/buzz-acp/src/acp.rs

@@ -214,6 +214,56 @@ pub struct AcpClient {
     standard_usage: StandardUsageTracker,
     /// Known adapter identity for prompt-response usage mapping.
     standard_adapter: Option<StandardAdapterKind>,
+    /// 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
@@ -563,6 +613,8 @@ impl AcpClient {
             goose_usage: UsageTracker::default(),
             standard_usage: StandardUsageTracker::default(),
             standard_adapter,
+            turn_text: String::new(),
+            turn_published: false,
         })
     }
 
@@ -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" => {
@@ -2352,6 +2444,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

@@ -428,6 +428,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(
@@ -629,6 +640,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.
@@ -1204,6 +1218,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)
@@ -1579,6 +1594,7 @@ mod tests {
             replay_floor_unix: None,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }

crates/buzz-acp/src/lib.rs

@@ -2825,6 +2825,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(),
     });
@@ -9100,6 +9101,7 @@ mod build_mcp_servers_tests {
             replay_floor_unix: None,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }
@@ -9326,6 +9328,7 @@ mod error_outcome_emission_tests {
             replay_floor_unix: None,
             agent_owner: None,
             no_base_prompt: false,
+            deliver_plain_replies: false,
             base_prompt_content: None,
         }
     }

crates/buzz-acp/src/pool.rs

@@ -805,6 +805,11 @@ pub struct PromptContext {
     /// `<core-memory>` 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,
@@ -2143,6 +2148,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!({
@@ -3093,6 +3106,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(
@@ -5054,6 +5092,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
@@ -5063,6 +5129,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()?;
@@ -5087,21 +5168,21 @@ pub(crate) async fn post_failure_notice(
     ) {
         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"),
     }
 }
 
@@ -8968,6 +9049,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             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(),
         }