diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8a698954a0..b1626d3f75 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/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::() + .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 { + 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)); diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index dab61be30a..5482f37314 100644 --- a/crates/buzz-acp/src/config.rs +++ b/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, + /// 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, } } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index d63f720c65..91bec83551 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/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, } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 038f8a714c..96f3317995 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/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 { + 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(), } diff --git a/desktop/src-tauri/src/managed_agents/relay_mesh.rs b/desktop/src-tauri/src/managed_agents/relay_mesh.rs index 327c106bc8..b28cff058f 100644 --- a/desktop/src-tauri/src/managed_agents/relay_mesh.rs +++ b/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]