diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index b553adaab..76ddae826 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -200,6 +200,17 @@ 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, + /// Accumulated `agent_message_chunk` text for the current turn. Used by the + /// content-delivery fallback: weak local models (e.g. via Buzz shared + /// compute) often answer a conversational prompt in plain assistant + /// `content` instead of calling `buzz messages send`, which would otherwise + /// be silently dropped (buzz-agent's output is its tool calls; streamed + /// text is observability-only). Reset at the start of every turn. + turn_message_text: String, + /// Whether a `buzz messages send` (or forum-post/comment) publish tool call + /// was observed this turn. When true the fallback does NOT fire — the agent + /// delivered its own reply. Reset at the start of every turn. + turn_sent_message: bool, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -492,6 +503,8 @@ impl AcpClient { active_run_id: None, steer_rx: None, goose_usage: UsageTracker::default(), + turn_message_text: String::new(), + turn_sent_message: false, }) } @@ -685,6 +698,10 @@ impl AcpClient { // misattributed to this turn. self.goose_usage.begin_turn(session_id); + // Reset the content-delivery fallback trackers for this turn. + self.turn_message_text.clear(); + self.turn_sent_message = false; + self.last_prompt_id = Some(self.next_id); let id = self.next_id; self.next_id += 1; @@ -780,6 +797,29 @@ impl AcpClient { self.goose_usage.take() } + /// Take the accumulated assistant `content` text for the completed turn, + /// if and only if the agent did NOT publish a message itself this turn. + /// + /// Returns `Some(trimmed_text)` when the turn produced streamed assistant + /// content but no `buzz messages send` tool call fired — the caller then + /// delivers it as the channel reply (content-delivery fallback). Returns + /// `None` when the agent sent its own message, when there was no content, + /// or when the content is only a bare acknowledgement (which the base + /// prompt forbids publishing). Clears the buffer either way. + pub fn take_undelivered_turn_message(&mut self) -> Option { + let text = std::mem::take(&mut self.turn_message_text); + let sent = self.turn_sent_message; + self.turn_sent_message = false; + if sent { + 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. /// @@ -1531,6 +1571,10 @@ impl AcpClient { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { tracing::info!(target: "acp::stream", "{text}"); + // Accumulate for the content-delivery fallback (see + // `turn_message_text`). Streamed assistant text is otherwise + // observability-only and never posted to the channel. + self.turn_message_text.push_str(text); } false } @@ -1544,6 +1588,17 @@ impl AcpClient { .and_then(|v| v.as_str()) .unwrap_or("unknown"); tracing::info!(target: "acp::tool", "tool_call: {title} ({kind})"); + // Detect a message-publish tool call so the fallback knows the + // agent delivered its own reply. The publish path is the + // dev-mcp `shell` tool running `buzz messages send` (title is + // the tool name, rawInput carries the command/args), so scan + // both title and rawInput for the CLI publish signature. + if tool_call_is_message_publish(update) { + self.turn_sent_message = true; + // Debug-level: the model published its own reply, so the + // content-delivery fallback will stay dormant this turn. + tracing::debug!("agent published its own message via send tool ({title})"); + } true } "tool_call_update" => { @@ -1944,6 +1999,73 @@ pub fn model_in_catalog( }) } +/// Return true if a `tool_call` session update represents a Buzz message +/// publish (kind 9 / forum post / comment). The publish path is the dev-mcp +/// `shell` tool running `buzz messages send` (or `buzz social publish`), so the +/// tool name alone is not enough — inspect `rawInput` (the command/args) for the +/// CLI publish signature. Conservative: only matches an actual send subcommand, +/// not reads like `buzz messages get`. +fn tool_call_is_message_publish(update: &serde_json::Value) -> bool { + // Flatten title + rawInput into one lowercase haystack. rawInput is + // arbitrary JSON (shell command string, or structured args), so serialize + // whatever is there. + let mut haystack = String::new(); + if let Some(title) = update.get("title").and_then(|v| v.as_str()) { + haystack.push_str(title); + haystack.push(' '); + } + if let Some(raw) = update.get("rawInput") { + haystack.push_str(&raw.to_string()); + } + let h = haystack.to_ascii_lowercase(); + // Match the publish subcommands that actually post to a channel. Guard + // against read subcommands (get/thread/search/list) sharing the "messages" + // prefix by requiring the send/publish verb. + h.contains("messages send") || h.contains("messages send-diff") || h.contains("social publish") +} + +/// Return true if `text` is a bare acknowledgement the base prompt forbids +/// publishing ("Got it", "Confirmed", "Standing by", …). Used to keep the +/// content-delivery fallback from posting filler that a capable agent would +/// have suppressed. Deliberately conservative — only short, whole-message +/// acks match, so a substantive reply that merely opens with "Got it, …" +/// still gets delivered. +fn is_bare_acknowledgement(text: &str) -> bool { + // Only consider short messages — a real reply with content is never a bare + // ack even if it starts with one. + if text.chars().count() > 40 { + return false; + } + let normalized: String = text + .to_ascii_lowercase() + .chars() + .filter(|c| c.is_alphanumeric() || c.is_whitespace()) + .collect(); + let normalized = normalized.trim(); + const BARE_ACKS: &[&str] = &[ + "got it", + "confirmed", + "acknowledged", + "ack", + "clear and noted", + "noted", + "aligned", + "standing by", + "parked", + "ok", + "okay", + "will do", + "understood", + "sounds good", + "on it", + "roger", + "roger that", + "i wont reply again", + "i will not reply again", + ]; + BARE_ACKS.contains(&normalized) +} + // ─── Drop: kill child process ───────────────────────────────────────────────── impl Drop for AcpClient { @@ -1991,6 +2113,70 @@ fn kill_process_group(_pid: u32) -> bool { mod tests { use super::*; + #[test] + fn tool_call_publish_detection() { + // A `buzz messages send` shell tool call → detected as a publish. + let send = serde_json::json!({ + "title": "shell", + "rawInput": { "command": "buzz messages send --channel abc --content 'hi'" } + }); + assert!(tool_call_is_message_publish(&send)); + + // send-diff variant → detected. + let diff = serde_json::json!({ + "title": "shell", + "rawInput": { "command": "buzz messages send-diff --channel abc" } + }); + assert!(tool_call_is_message_publish(&diff)); + + // social publish → detected. + let social = serde_json::json!({ + "title": "shell", + "rawInput": { "command": "buzz social publish --content x" } + }); + assert!(tool_call_is_message_publish(&social)); + + // A READ subcommand sharing the "messages" prefix → NOT a publish. + let read = serde_json::json!({ + "title": "shell", + "rawInput": { "command": "buzz messages get --channel abc" } + }); + assert!(!tool_call_is_message_publish(&read)); + + // Unrelated tool → not a publish. + let other = serde_json::json!({ + "title": "read_file", + "rawInput": { "path": "/tmp/foo" } + }); + assert!(!tool_call_is_message_publish(&other)); + } + + #[test] + fn bare_acknowledgement_detection() { + // Bare acks the base prompt forbids publishing. + for ack in [ + "Got it", + "confirmed", + "Standing by", + "OK", + " Noted. ", + "will do", + ] { + assert!(is_bare_acknowledgement(ack), "should be bare ack: {ack:?}"); + } + // Substantive replies are NOT bare acks, even if they open with one. + for real in [ + "Got it — I'll start on the migration and report back when the tests pass.", + "I'm doing well, thank you for asking! How are you today?", + "The build failed: missing dependency in Cargo.toml.", + ] { + assert!( + !is_bare_acknowledgement(real), + "should NOT be bare ack: {real:?}" + ); + } + } + #[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/pool.rs b/crates/buzz-acp/src/pool.rs index fb0809679..6243902a9 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1775,6 +1775,28 @@ pub async fn run_prompt_task( None => prompt_sections.iter().map(String::as_str).collect(), }; + // Capture the reply destination for the content-delivery fallback BEFORE + // the prompt runs, so it survives any move of `batch` in the outcome arms. + // Only channel turns with a triggering event can receive a fallback post; + // heartbeats and DMs without a triggering message are skipped (None). + let fallback_reply: Option = batch.as_ref().and_then(|b| { + b.events.last().map(|last| { + let tags = crate::queue::parse_thread_tags(&last.event); + // Thread the reply to the triggering event: if the trigger is + // itself a reply, anchor to its root; otherwise the trigger IS + // the root. Mirrors the CLI `resolve_thread_ref` semantics. + let root_hex = tags + .root_event_id + .clone() + .unwrap_or_else(|| last.event.id.to_hex()); + FallbackReplyTarget { + channel_id: b.channel_id, + root_event_hex: root_hex, + parent_event_hex: last.event.id.to_hex(), + } + }) + }); + // When control_rx is Some (channel tasks), wrap the prompt in select! so // the main loop can cancel, interrupt, or rotate it. Heartbeats // (control_rx=None) take the simple await path — they are not controllable. @@ -1927,6 +1949,15 @@ pub async fn run_prompt_task( Some(buzz_core::agent_turn_metric::StopReason::EndTurn), ) .await; + // Content-delivery fallback (see the main EndTurn arm): + // this rare branch is also a successful turn end, so an + // undelivered plain-text reply still needs posting. + if let (Some(target), Some(content)) = + (&fallback_reply, agent.acp.take_undelivered_turn_message()) + { + post_agent_content_fallback(&ctx.rest_client, target, &content) + .await; + } send_prompt_result( &result_tx, &turn_id, @@ -1990,6 +2021,21 @@ pub async fn run_prompt_task( ) .await; + // Content-delivery fallback: on a normal turn end, if the agent + // produced assistant text but never called a publish tool, post + // that text as the channel reply. Only fires for `EndTurn` (not + // MaxTokens/MaxTurnRequests, which are truncated/aborted turns + // whose partial text shouldn't be treated as a deliberate reply) + // and only when a `fallback_reply` destination was captured + // (channel turns with a triggering event; not heartbeats). + if matches!(stop_reason, StopReason::EndTurn) { + if let (Some(target), Some(content)) = + (&fallback_reply, agent.acp.take_undelivered_turn_message()) + { + post_agent_content_fallback(&ctx.rest_client, target, &content).await; + } + } + send_prompt_result( &result_tx, &turn_id, @@ -3495,6 +3541,81 @@ pub(crate) async fn post_failure_notice( } } +/// Captured reply destination for the content-delivery fallback, taken before +/// the prompt runs so it survives any move of the triggering `batch`. +#[derive(Clone)] +struct FallbackReplyTarget { + channel_id: Uuid, + /// Thread root the reply anchors to (hex). Equals `parent_event_hex` when + /// the trigger was a top-level message. + root_event_hex: String, + /// Immediate parent being replied to (hex) — the triggering event. + parent_event_hex: String, +} + +/// Content-delivery fallback: post an agent's plain-text reply (kind:9) that it +/// generated but never published itself. +/// +/// buzz-agent's output is its tool calls; streamed assistant `content` is +/// observability-only and is normally not posted. Capable models reliably call +/// `buzz messages send`, but weaker local models (e.g. via Buzz shared compute) +/// often answer a conversational prompt in plain content and never call the +/// send tool — silently dropping the reply. When [`AcpClient`] reports such an +/// undelivered turn message, this posts it as a threaded reply, mirroring +/// [`post_failure_notice`]'s build/sign/submit path. Best-effort: any error is +/// logged and swallowed. +async fn post_agent_content_fallback( + rest: &crate::relay::RestClient, + target: &FallbackReplyTarget, + content: &str, +) { + let thread_ref = match ( + nostr::EventId::from_hex(&target.root_event_hex), + nostr::EventId::from_hex(&target.parent_event_hex), + ) { + (Ok(root_id), Ok(parent_id)) => Some(buzz_sdk::ThreadRef { + root_event_id: root_id, + parent_event_id: parent_id, + }), + _ => None, + }; + let builder = match buzz_sdk::build_message( + target.channel_id, + content, + thread_ref.as_ref(), + &[], + false, + &[], + ) { + Ok(b) => b, + Err(e) => { + tracing::warn!(channel = %target.channel_id, "content fallback: build failed: {e}"); + return; + } + }; + let event = match builder.sign_with_keys(&rest.keys) { + Ok(e) => e, + Err(e) => { + tracing::warn!(channel = %target.channel_id, "content fallback: sign failed: {e}"); + return; + } + }; + match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await { + Ok(Ok(_)) => { + // WARN (not INFO) and default target (buzz_acp::pool) so it is + // always visible under the harness's `buzz_acp=info` filter — this + // fallback firing is a signal worth surfacing (a model failed to + // call the send tool and we delivered its reply for it). + tracing::warn!( + channel = %target.channel_id, + "content-delivery fallback: posted undelivered agent content as channel reply" + ); + } + Ok(Err(e)) => tracing::warn!(channel = %target.channel_id, "content fallback failed: {e}"), + Err(_) => tracing::warn!(channel = %target.channel_id, "content fallback timed out"), + } +} + /// Best-effort: remove a reaction via a signed kind:5 (NIP-09) deletion event. /// /// Queries kind:7 reactions by our pubkey targeting the event, finds the matching