From bca8bab78a71ab0d78b71e90f2723a1f0199b10f Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Thu, 23 Jul 2026 15:29:15 +1000 Subject: [PATCH 1/4] fix(acp): deliver plain-text replies when the model skips the send tool buzz-agent's output is its tool calls; streamed assistant content is observability-only and normally never 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. Add a content-delivery fallback in buzz-acp: track per-turn whether the model published its own message and buffer streamed content; on a normal turn end with content but no publish tool call, post that content as a threaded kind-9 reply (reusing buzz_sdk::build_message, signed with the agent keys, best-effort, mirroring post_failure_notice). Skips bare acknowledgements the base prompt forbids publishing. Live-validated against a small Gemma on a real community: the fallback fires when the model answers in prose and stays dormant when it calls the send tool. Unit tests cover send-tool detection (incl. read-vs-send discrimination) and bare-ack filtering. (cherry picked from commit 1fe074bc73b4c8d153aba8d7e2a96522a4e76636) Co-authored-by: chuck Signed-off-by: chuck --- crates/buzz-acp/src/acp.rs | 186 ++++++++++++++++++++++++++++++++++++ crates/buzz-acp/src/pool.rs | 123 +++++++++++++++++++++++- 2 files changed, 308 insertions(+), 1 deletion(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8a698954a0..9a7b7270e9 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -211,6 +211,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 @@ -550,6 +561,8 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), + turn_message_text: String::new(), + turn_sent_message: false, }) } @@ -760,6 +773,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; @@ -864,6 +881,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. /// @@ -1715,6 +1755,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 } @@ -1728,6 +1772,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" => { @@ -2163,6 +2218,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 { @@ -2223,6 +2345,70 @@ fn configure_no_window(cmd: &mut tokio::process::Command) { 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 348bc138e4..f0f67bc7a6 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1901,7 +1901,7 @@ pub async fn run_prompt_task( None => prompt_sections.iter().map(String::as_str).collect(), }; - // Turn start, labelled exactly as `log_stop_reason` labels the end, so a +// Turn start, labelled exactly as `log_stop_reason` labels the end, so a // log reads as start/stop pairs. Purely observational: an unpaired start is // the only durable evidence that a turn was entered and never returned, and // without it a stalled agent and an agent nobody woke leave identical logs — @@ -1913,6 +1913,28 @@ pub async fn run_prompt_task( prompt_label(&source) ); + // 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. @@ -2065,6 +2087,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, @@ -2128,6 +2159,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, @@ -3854,6 +3900,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 From f68a1d7aea0ea1a6cf8be8f95b720ba2df20773f Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Thu, 23 Jul 2026 21:09:41 -0700 Subject: [PATCH 2/4] fix(acp): confirm message delivery from tool outcome, not input text MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings from #2448 (Wren + Eva): the content-delivery fallback inferred delivery from the send tool call's INPUT, which was wrong in both directions: - False negative (silent drop): turn_sent_message was set the moment a tool_call's rawInput matched 'messages send' — intent, not delivery. A send that subsequently failed (relay/network/auth/CLI error) still suppressed the fallback, dropping the exact reply the feature exists to save. - False positive (double post): the substring match missed argv-style publishes (subprocess.run(['buzz','messages','send',...])), so a capable model that narrates between tool calls and publishes via argv would get its whole turn narration re-posted by the fallback. Fix: registering a publish is now two-phase. - tool_call input matching (normalized so shell strings AND argv forms match) only registers a CANDIDATE toolCallId. - Delivery is confirmed solely from the candidate's terminal tool_call_update outcome: failed/cancelled → not delivered (fallback stays armed); completed → inspect rawOutput.isError, the CLI's {"accepted":...} response envelope, and any reported exit_code; a bare completed with no inspectable output keeps the status-quo direction (suppress, i.e. today's behavior). Tests: outcome classifier matrix (pending/failed/cancelled/completed x envelope/isError/exit_code) plus handler-level lifecycle tests through handle_session_update + take_undelivered_turn_message covering success, failure, cancel, failed-then-retry, and argv-style publish. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell (cherry picked from commit e992ae745d7d6238b3b25f8b6c2d03c7058c966b) --- crates/buzz-acp/src/acp.rs | 457 ++++++++++++++++++++++++++++++++++--- 1 file changed, 430 insertions(+), 27 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 9a7b7270e9..46fffa24a3 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -218,10 +218,20 @@ pub struct AcpClient { /// 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. + /// Whether a message publish was CONFIRMED this turn. When true the + /// fallback does NOT fire — the agent delivered its own reply. Confirmation + /// comes from the terminal `tool_call_update` outcome (successful + /// completion, ideally carrying the CLI's `{"accepted":true,...}` envelope + /// in the tool output), never from the tool call's input text alone: + /// intent is not delivery. Reset at the start of every turn. turn_sent_message: bool, + /// Publish *candidates* for the current turn: `toolCallId`s whose + /// `tool_call` input matched the publish signature, awaiting a terminal + /// `tool_call_update`. A candidate that completes successfully confirms + /// delivery; one that fails (or completes with `isError`, or whose output + /// lacks the publish acknowledgement) is discarded so the fallback stays + /// armed. Reset at the start of every turn. + turn_publish_candidates: std::collections::HashSet, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -563,6 +573,7 @@ impl AcpClient { goose_usage: UsageTracker::default(), turn_message_text: String::new(), turn_sent_message: false, + turn_publish_candidates: std::collections::HashSet::new(), }) } @@ -776,6 +787,7 @@ impl AcpClient { // Reset the content-delivery fallback trackers for this turn. self.turn_message_text.clear(); self.turn_sent_message = false; + self.turn_publish_candidates.clear(); self.last_prompt_id = Some(self.next_id); let id = self.next_id; @@ -882,14 +894,15 @@ impl AcpClient { } /// Take the accumulated assistant `content` text for the completed turn, - /// if and only if the agent did NOT publish a message itself this turn. + /// if and only if a message publish was NOT confirmed 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. + /// content but no publish tool call was confirmed delivered (see + /// `publish_outcome_confirms_delivery`) — the caller then delivers it as + /// the channel reply (content-delivery fallback). Returns `None` when the + /// agent's own send was confirmed, 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; @@ -1772,16 +1785,27 @@ 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. + // Register a message-publish CANDIDATE for the content-delivery + // fallback. Intent is not delivery: the flag that suppresses + // the fallback (`turn_sent_message`) is only set when this + // call's terminal `tool_call_update` confirms success — a send + // that fails must leave the fallback armed, otherwise the + // feature silently drops the exact reply it exists to save. 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})"); + if let Some(id) = update.get("toolCallId").and_then(|v| v.as_str()) { + self.turn_publish_candidates.insert(id.to_string()); + tracing::debug!( + "publish candidate registered ({title}, toolCallId={id}); \ + awaiting terminal outcome" + ); + } + // Some agents emit `tool_call` already carrying a terminal + // status (single-event shape). Handle it like an update. + if let Some(confirmed) = publish_outcome_confirms_delivery(update) { + if confirmed { + self.turn_sent_message = true; + } + } } true } @@ -1792,6 +1816,24 @@ impl AcpClient { .unwrap_or("?"); let status = update.get("status").and_then(|v| v.as_str()).unwrap_or("?"); tracing::info!(target: "acp::tool", "tool_call_update: {tool_id} → {status}"); + // Resolve a pending publish candidate on its terminal outcome. + if self.turn_publish_candidates.contains(tool_id) { + if let Some(confirmed) = publish_outcome_confirms_delivery(update) { + self.turn_publish_candidates.remove(tool_id); + if confirmed { + self.turn_sent_message = true; + tracing::debug!( + "publish confirmed (toolCallId={tool_id}); \ + content-delivery fallback stays dormant this turn" + ); + } else { + tracing::debug!( + "publish attempt failed (toolCallId={tool_id}); \ + content-delivery fallback stays armed" + ); + } + } + } false } "plan" => { @@ -2218,12 +2260,19 @@ 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`. +/// Return true if a `tool_call` session update *looks like* a Buzz message +/// publish (kind 9 / forum post / comment) from its input. This only registers +/// a publish **candidate** — delivery is confirmed separately from the call's +/// terminal outcome by [`publish_outcome_confirms_delivery`]. +/// +/// 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. The haystack is +/// normalized (quotes/commas/brackets → spaces, whitespace collapsed) so both +/// shell strings (`buzz messages send --channel …`) and argv forms +/// (`['buzz','messages','send',…]`, e.g. Python `subprocess.run`) match. +/// 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 @@ -2236,11 +2285,112 @@ fn tool_call_is_message_publish(update: &serde_json::Value) -> bool { if let Some(raw) = update.get("rawInput") { haystack.push_str(&raw.to_string()); } - let h = haystack.to_ascii_lowercase(); + // Normalize away quoting/punctuation so argv-style invocations + // ('buzz','messages','send') match the same signature as shell strings. + let normalized: String = haystack + .to_ascii_lowercase() + .chars() + .map(|c| match c { + '\'' | '"' | '`' | ',' | '[' | ']' | '(' | ')' | '{' | '}' | ':' => ' ', + other => other, + }) + .collect(); + let h = normalized.split_whitespace().collect::>().join(" "); // 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") + // prefix by requiring the send/publish verb. ("messages send-diff" + // contains "messages send", so it is covered.) + h.contains("messages send") || h.contains("social publish") +} + +/// Classify the terminal outcome of a publish tool call. +/// +/// Returns `None` while the call is still pending/in-progress, `Some(true)` +/// when the outcome confirms the message was delivered, and `Some(false)` when +/// the attempt failed (so the content-delivery fallback must stay armed — +/// see the `tool_call`/`tool_call_update` arms in `handle_session_update`). +/// +/// Signals, strongest first: +/// 1. `status: "failed"` (or cancelled) → not delivered. +/// 2. `rawOutput.isError: true` → not delivered (buzz-agent's builtin shape). +/// 3. Visible tool output containing the CLI's response envelope: +/// `"accepted":true` confirms, `"accepted":false` denies. +/// 4. A reported `exit_code` in the output (dev-mcp `shell` completes the +/// *tool* call even when the *command* failed): nonzero → not delivered. +/// 5. Otherwise, a `completed` non-error publish attempt counts as delivered — +/// the status-quo direction (suppressed fallback == today's behavior), +/// chosen over risking a duplicate post when output isn't visible. +fn publish_outcome_confirms_delivery(update: &serde_json::Value) -> Option { + let status = update.get("status").and_then(|v| v.as_str())?; + match status { + "failed" | "cancelled" | "canceled" | "error" => Some(false), + "completed" => { + if update + .get("rawOutput") + .and_then(|r| r.get("isError")) + .and_then(serde_json::Value::as_bool) + == Some(true) + { + return Some(false); + } + // Compact the visible output (content blocks + rawOutput) so the + // envelope matches regardless of pretty-printing — and strip + // backslashes so JSON nested inside a JSON string (rawOutput + // serialization escapes the quotes) matches too. + let compact: String = publish_output_text(update) + .chars() + .filter(|c| !c.is_whitespace() && *c != '\\') + .collect(); + if compact.contains(r#""accepted":true"#) { + return Some(true); + } + if compact.contains(r#""accepted":false"#) { + return Some(false); + } + if let Some(code) = extract_reported_exit_code(&compact) { + return Some(code == 0); + } + Some(true) + } + // "pending" / "in_progress" / anything non-terminal. + _ => None, + } +} + +/// Gather the human-visible output of a tool call update: ACP `content` text +/// blocks plus the serialized `rawOutput`, whichever are present. +fn publish_output_text(update: &serde_json::Value) -> String { + let mut out = String::new(); + if let Some(items) = update.get("content").and_then(|c| c.as_array()) { + for item in items { + // ACP shape: {type:"content", content:{type:"text", text:…}}; + // tolerate a flat {text:…} too. + if let Some(t) = item.pointer("/content/text").and_then(|v| v.as_str()) { + out.push_str(t); + out.push(' '); + } else if let Some(t) = item.get("text").and_then(|v| v.as_str()) { + out.push_str(t); + out.push(' '); + } + } + } + if let Some(raw) = update.get("rawOutput") { + out.push_str(&raw.to_string()); + } + out +} + +/// Extract a `"exit_code": N` value from compacted (whitespace-free) tool +/// output, e.g. the dev-mcp `shell` tool's result JSON. Returns `None` when no +/// exit code is reported. +fn extract_reported_exit_code(compact: &str) -> Option { + const KEY: &str = r#""exit_code":"#; + let idx = compact.find(KEY)?; + let rest = &compact[idx + KEY.len()..]; + let end = rest + .find(|c: char| !(c.is_ascii_digit() || c == '-')) + .unwrap_or(rest.len()); + rest[..end].parse().ok() } /// Return true if `text` is a bare acknowledgement the base prompt forbids @@ -2368,6 +2518,17 @@ mod tests { }); assert!(tool_call_is_message_publish(&social)); + // Python-argv publish (the pattern agents use for backtick-heavy + // content): serialized rawInput has no "messages send" substring, but + // normalization must still match it. + let argv = serde_json::json!({ + "title": "shell", + "rawInput": { + "command": "python3 - <<'PY'\nimport subprocess\nsubprocess.run(['buzz','messages','send','--channel','abc','--content',content])\nPY" + } + }); + assert!(tool_call_is_message_publish(&argv)); + // A READ subcommand sharing the "messages" prefix → NOT a publish. let read = serde_json::json!({ "title": "shell", @@ -2383,6 +2544,88 @@ mod tests { assert!(!tool_call_is_message_publish(&other)); } + #[test] + fn publish_outcome_classification() { + // Non-terminal statuses → None (candidate stays pending). + for status in ["pending", "in_progress"] { + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ "status": status })), + None, + "{status} is not terminal" + ); + } + // No status at all (e.g. a content-only update) → None. + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({})), + None + ); + + // Failure statuses → Some(false): the fallback must stay armed. + for status in ["failed", "cancelled", "canceled", "error"] { + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ "status": status })), + Some(false), + "{status} must not confirm delivery" + ); + } + + // completed + rawOutput.isError → not delivered (buzz-agent shape). + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ + "status": "completed", + "rawOutput": { "isError": true } + })), + Some(false) + ); + + // completed + CLI envelope accepted:true in the content text → delivered. + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ + "status": "completed", + "content": [{ "type": "content", "content": { "type": "text", + "text": "0 {\"accepted\": true, \"event_id\": \"abc\"}" } }] + })), + Some(true) + ); + + // completed + envelope accepted:false (relay rejected) → not delivered. + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ + "status": "completed", + "rawOutput": { "stdout": "{\"accepted\": false, \"message\": \"rate limited\"}" } + })), + Some(false) + ); + + // completed, no envelope, dev-mcp shell reports nonzero exit_code → + // the COMMAND failed even though the TOOL completed. Not delivered. + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ + "status": "completed", + "content": [{ "type": "content", "content": { "type": "text", + "text": "{\"exit_code\": 3, \"stderr\": \"auth failure\"}" } }] + })), + Some(false) + ); + + // completed, exit_code 0, no envelope → delivered. + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ + "status": "completed", + "content": [{ "type": "content", "content": { "type": "text", + "text": "{\"exit_code\": 0, \"stdout\": \"sent\"}" } }] + })), + Some(true) + ); + + // completed with no inspectable output at all → delivered (status-quo + // direction: suppressing the fallback == today's behavior). + assert_eq!( + publish_outcome_confirms_delivery(&serde_json::json!({ "status": "completed" })), + Some(true) + ); + } + #[test] fn bare_acknowledgement_detection() { // Bare acks the base prompt forbids publishing. @@ -2409,6 +2652,166 @@ mod tests { } } + /// Build a `session/update` notification wrapping the given update object. + fn session_update_msg(update: serde_json::Value) -> serde_json::Value { + serde_json::json!({ "params": { "update": update } }) + } + + /// Successful send: candidate registered on `tool_call`, confirmed on the + /// terminal completed update with the CLI envelope → fallback suppressed. + #[tokio::test] + async fn fallback_suppressed_when_send_completes_successfully() { + let mut client = spawn_inert_client().await; + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "Here's my full reply narration." } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-1", + "title": "shell", + "status": "pending", + "rawInput": { "command": "buzz messages send --channel abc --content 'hi'" } + }))); + // Not yet confirmed: a crash here must leave the fallback ARMED. + assert!(!client.turn_sent_message); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-1", + "status": "completed", + "rawOutput": { "stdout": "{\"accepted\":true,\"event_id\":\"e1\"}" } + }))); + assert!( + client.turn_sent_message, + "successful send must confirm delivery" + ); + assert_eq!( + client.take_undelivered_turn_message(), + None, + "confirmed delivery suppresses the fallback" + ); + } + + /// Failed send: candidate registered, terminal update is `failed` → the + /// fallback stays armed and the buffered content is released for posting. + /// This is the false-negative path from review: intent is not delivery. + #[tokio::test] + async fn fallback_stays_armed_when_send_fails() { + let mut client = spawn_inert_client().await; + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "The answer is 42." } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-2", + "title": "shell", + "status": "pending", + "rawInput": { "command": "buzz messages send --channel abc --content 'The answer is 42.'" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-2", + "status": "failed", + "rawOutput": { "error": "relay unreachable" } + }))); + assert!( + !client.turn_sent_message, + "failed send must NOT count as delivery" + ); + assert_eq!( + client.take_undelivered_turn_message().as_deref(), + Some("The answer is 42."), + "failed send leaves the fallback armed with the buffered reply" + ); + } + + /// Cancelled send behaves like failure: fallback stays armed. + #[tokio::test] + async fn fallback_stays_armed_when_send_cancelled() { + let mut client = spawn_inert_client().await; + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "Reply that never made it out." } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-3", + "title": "shell", + "status": "pending", + "rawInput": { "command": "buzz messages send --channel abc --content x" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-3", + "status": "cancelled" + }))); + assert!(!client.turn_sent_message); + assert!(client.take_undelivered_turn_message().is_some()); + } + + /// A failed attempt followed by a successful retry (new toolCallId) + /// confirms delivery — the fallback must not double-post after a retry. + #[tokio::test] + async fn fallback_suppressed_after_failed_then_successful_retry() { + let mut client = spawn_inert_client().await; + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "narration" } + }))); + for (id, status) in [("tc-4a", "failed"), ("tc-4b", "completed")] { + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": id, + "title": "shell", + "status": "pending", + "rawInput": { "command": "buzz messages send --channel abc --content x" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": id, + "status": status, + "rawOutput": { "stdout": "{\"accepted\":true,\"event_id\":\"e2\"}" } + }))); + } + assert!( + client.turn_sent_message, + "retry succeeded — delivery confirmed" + ); + assert_eq!(client.take_undelivered_turn_message(), None); + } + + /// Argv-style publish (Python subprocess) is recognized as a candidate and + /// confirmed on success — the normalization regression from review. + #[tokio::test] + async fn fallback_suppressed_for_argv_style_publish() { + let mut client = spawn_inert_client().await; + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "long narration between tool calls" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-5", + "title": "shell", + "status": "pending", + "rawInput": { "command": "python3 - <<'PY'\nimport subprocess\nsubprocess.run(['buzz','messages','send','--channel','abc','--content',content])\nPY" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-5", + "status": "completed", + "content": [{ "type": "content", "content": { "type": "text", + "text": "0 {\"accepted\": true, \"event_id\": \"abc\"}" } }] + }))); + assert!(client.turn_sent_message); + assert_eq!( + client.take_undelivered_turn_message(), + None, + "argv publish must not double-post the narration" + ); + } + #[test] fn stop_reason_parses_all_known_values() { assert_eq!(StopReason::from_str("end_turn"), Some(StopReason::EndTurn)); From f5e5ecb897bfae2f8b03ef699f106ab9d9730a69 Mon Sep 17 00:00:00 2001 From: chuck Date: Sat, 1 Aug 2026 18:34:37 -0400 Subject: [PATCH 3/4] feat(acp): S1 detect MCP buzz_messages_send tool names Port already on branch (plaintext fallback + outcome confirmation). Extend publish detection for goose-prefixed MCP tool ids so Anvil's buzz_publish__buzz_messages_send path still suppresses double-post. plan(p2-m2) Co-authored-by: chuck Signed-off-by: chuck --- crates/buzz-acp/src/acp.rs | 54 +++++++++++++++++++++++++++++++++----- 1 file changed, 47 insertions(+), 7 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 46fffa24a3..3beead44c3 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -2265,14 +2265,18 @@ pub fn model_in_catalog( /// a publish **candidate** — delivery is confirmed separately from the call's /// terminal outcome by [`publish_outcome_confirms_delivery`]. /// -/// 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. The haystack is -/// normalized (quotes/commas/brackets → spaces, whitespace collapsed) so both -/// shell strings (`buzz messages send --channel …`) and argv forms +/// The publish path is either: +/// - the dev-mcp `shell` tool running `buzz messages send` / `buzz social publish` +/// - a first-class MCP tool whose id ends in / is `buzz_messages_send` +/// (goose prefixes extension names: `buzz_publish__buzz_messages_send`, +/// `buzz-dev-mcp__buzz_messages_send`) +/// +/// Inspect title + `rawInput` for those signatures. The haystack is normalized +/// (quotes/commas/brackets → spaces, whitespace collapsed) so both shell strings +/// (`buzz messages send --channel …`) and argv forms /// (`['buzz','messages','send',…]`, e.g. Python `subprocess.run`) match. /// Conservative: only matches an actual send subcommand, not reads like -/// `buzz messages get`. +/// `buzz messages get` or `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 @@ -2300,7 +2304,13 @@ fn tool_call_is_message_publish(update: &serde_json::Value) -> bool { // against read subcommands (get/thread/search/list) sharing the "messages" // prefix by requiring the send/publish verb. ("messages send-diff" // contains "messages send", so it is covered.) - h.contains("messages send") || h.contains("social publish") + // + // MCP tool ids use underscores (and optional goose `ext__` prefixes), so + // also match the token `buzz_messages_send` — JSON rawInput alone has no + // "messages send" substring. + h.contains("messages send") + || h.contains("social publish") + || h.contains("buzz_messages_send") } /// Classify the terminal outcome of a publish tool call. @@ -2544,6 +2554,36 @@ mod tests { assert!(!tool_call_is_message_publish(&other)); } + #[test] + fn tool_call_mcp_buzz_messages_send_detected() { + // Goose-prefixed MCP publish tool: title carries buzz_messages_send; + // rawInput is structured JSON without the words "messages send". + let update = serde_json::json!({ + "title": "buzz_publish__buzz_messages_send", + "rawInput": { "channel": "aa83e87b-f7da-4014-83ed-824a6ff3a4d0", "content": "hi" } + }); + assert!(tool_call_is_message_publish(&update)); + + let bare = serde_json::json!({ + "title": "buzz_messages_send", + "rawInput": { "channel": "x", "content": "y" } + }); + assert!(tool_call_is_message_publish(&bare)); + + let alt_prefix = serde_json::json!({ + "title": "buzz-dev-mcp__buzz_messages_send", + "rawInput": { "channel": "x", "content": "y", "reply_to": "abc" } + }); + assert!(tool_call_is_message_publish(&alt_prefix)); + + // Read MCP tool sharing the buzz_messages_ prefix → NOT a publish. + let read_only = serde_json::json!({ + "title": "buzz_publish__buzz_messages_get", + "rawInput": { "channel": "x" } + }); + assert!(!tool_call_is_message_publish(&read_only)); + } + #[test] fn publish_outcome_classification() { // Non-terminal statuses → None (candidate stays pending). From 1d0708a9a0e984d064335ce499a4c6342fc391c5 Mon Sep 17 00:00:00 2001 From: chuck Date: Sun, 2 Aug 2026 07:32:07 -0400 Subject: [PATCH 4/4] feat(acp): once-publish-per-trigger stop after first confirmed send After the first confirmed successful channel publish in a turn, send session/cancel so multi-step models cannot re-call buzz_messages_send (Cipher triple-post / Glitch B). Intentional cancel maps to EndTurn so the trigger batch is not requeued. Failed first publishes still allow retries until one confirms. Co-authored-by: chuck Signed-off-by: chuck --- crates/buzz-acp/src/acp.rs | 196 ++++++++++++++++++++++++++++++++++++- 1 file changed, 191 insertions(+), 5 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 3beead44c3..7115ee92a8 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -232,6 +232,21 @@ pub struct AcpClient { /// lacks the publish acknowledgement) is discarded so the fallback stays /// armed. Reset at the start of every turn. turn_publish_candidates: std::collections::HashSet, + /// Once-publish-per-trigger stop (Glitch B): set when the first channel + /// publish of this turn is **confirmed** delivered. The idle/simple read + /// loops send a single `session/cancel` so the model cannot multi-step + /// additional `buzz_messages_send` calls after tool success (Cipher + /// triple-post). Live agents often run `permission_mode=bypassPermissions`, + /// so a permission-reject gate is not available. Reset each turn. + once_publish_stop_pending: bool, + /// True after we have already sent `session/cancel` for once-publish this + /// turn (idempotent — avoid double cancel). + once_publish_stop_fired: bool, + /// True when the in-flight cancel was raised by once-publish. When the + /// agent returns `stopReason: cancelled`, we rewrite it to `end_turn` so + /// the pool treats the trigger batch as successfully processed (no + /// requeue). Reset each turn. + once_publish_stop_applied: bool, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -574,6 +589,9 @@ impl AcpClient { turn_message_text: String::new(), turn_sent_message: false, turn_publish_candidates: std::collections::HashSet::new(), + once_publish_stop_pending: false, + once_publish_stop_fired: false, + once_publish_stop_applied: false, }) } @@ -788,6 +806,10 @@ impl AcpClient { self.turn_message_text.clear(); self.turn_sent_message = false; self.turn_publish_candidates.clear(); + // Reset once-publish-per-trigger stop (Glitch B) for this turn. + self.once_publish_stop_pending = false; + self.once_publish_stop_fired = false; + self.once_publish_stop_applied = false; self.last_prompt_id = Some(self.next_id); let id = self.next_id; @@ -1269,6 +1291,9 @@ impl AcpClient { match method { "session/update" => { let _ = self.handle_session_update(&msg); + // Once-publish stop needs a session_id; simple + // read_until_response has none — fire only from the + // idle-timeout prompt loop (session_prompt path). } "_goose/unstable/session/update" => { self.handle_goose_usage_update(&msg); @@ -1713,6 +1738,11 @@ impl AcpClient { last_activity_at = activity_now; tracing::debug!("idle clock reset: tool call started"); } + // Glitch B: after first confirmed channel + // publish this turn, cancel so the model + // cannot multi-step additional publishes. + self.maybe_fire_once_publish_stop(session_id) + .await?; } "_goose/unstable/session/update" => { self.handle_goose_usage_update(&msg); @@ -1803,7 +1833,7 @@ impl AcpClient { // status (single-event shape). Handle it like an update. if let Some(confirmed) = publish_outcome_confirms_delivery(update) { if confirmed { - self.turn_sent_message = true; + self.mark_publish_confirmed(title); } } } @@ -1821,7 +1851,7 @@ impl AcpClient { if let Some(confirmed) = publish_outcome_confirms_delivery(update) { self.turn_publish_candidates.remove(tool_id); if confirmed { - self.turn_sent_message = true; + self.mark_publish_confirmed(tool_id); tracing::debug!( "publish confirmed (toolCallId={tool_id}); \ content-delivery fallback stays dormant this turn" @@ -2036,13 +2066,73 @@ impl AcpClient { Ok(()) } + /// Mark a channel/DM publish as confirmed delivered this turn and arm the + /// once-publish-per-trigger stop so the read loop will cancel further + /// multi-step agent work after the first successful send. + /// + /// Idempotent for the fallback flag (`turn_sent_message`); only the first + /// confirmation sets `once_publish_stop_pending` so a successful retry + /// after a failed attempt still cancels once (not zero times, not twice). + fn mark_publish_confirmed(&mut self, label: &str) { + let first = !self.turn_sent_message; + self.turn_sent_message = true; + if first { + self.once_publish_stop_pending = true; + tracing::info!( + target: "acp::once_publish", + "once-publish stop armed after first confirmed publish ({label}); \ + will cancel turn to prevent multi-step re-publish" + ); + } + } + + /// If once-publish stop is pending and we have not already cancelled, + /// send `session/cancel` for the in-flight prompt. Called from the + /// session/update read loops immediately after a confirmed publish. + async fn maybe_fire_once_publish_stop( + &mut self, + session_id: &str, + ) -> Result<(), AcpError> { + if !self.once_publish_stop_pending || self.once_publish_stop_fired { + return Ok(()); + } + if !self.has_in_flight_prompt() { + // Turn already completed (race): nothing to cancel. + self.once_publish_stop_pending = false; + return Ok(()); + } + self.session_cancel(session_id).await?; + self.once_publish_stop_fired = true; + self.once_publish_stop_applied = true; + self.once_publish_stop_pending = false; + tracing::info!( + target: "acp::once_publish", + "once-publish stop: sent session/cancel after first confirmed channel publish" + ); + Ok(()) + } + /// Parse `stopReason` from a `session/prompt` result value. - fn parse_stop_reason(&self, result: &serde_json::Value) -> Result { + /// + /// When the cancel was raised by once-publish-per-trigger, rewrite + /// `cancelled` → `end_turn` so the pool does not requeue the trigger batch + /// (Cancelled fate) after a successful first publish. + fn parse_stop_reason(&mut self, result: &serde_json::Value) -> Result { let raw = result["stopReason"].as_str().ok_or_else(|| { AcpError::Protocol("session/prompt response missing stopReason".into()) })?; - StopReason::from_str(raw) - .ok_or_else(|| AcpError::Protocol(format!("unknown stopReason: {raw:?}"))) + let reason = StopReason::from_str(raw) + .ok_or_else(|| AcpError::Protocol(format!("unknown stopReason: {raw:?}")))?; + if matches!(reason, StopReason::Cancelled) && self.once_publish_stop_applied { + tracing::info!( + target: "acp::once_publish", + "once-publish stop: rewriting stopReason cancelled → end_turn \ + (first publish already delivered; do not requeue trigger)" + ); + self.once_publish_stop_applied = false; + return Ok(StopReason::EndTurn); + } + Ok(reason) } } @@ -2821,6 +2911,102 @@ mod tests { assert_eq!(client.take_undelivered_turn_message(), None); } + /// Once-publish stop arms only on first *confirmed* publish, not on fail. + #[tokio::test] + async fn once_publish_stop_arms_on_first_confirmed_only() { + let mut client = spawn_inert_client().await; + assert!(!client.once_publish_stop_pending); + // Failed publish: no stop. + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-op-fail", + "title": "buzz_publish__buzz_messages_send", + "status": "pending", + "rawInput": { "content": "hi" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-op-fail", + "status": "failed" + }))); + assert!(!client.turn_sent_message); + assert!(!client.once_publish_stop_pending); + + // Successful publish: arm stop. + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-op-ok", + "title": "buzz_publish__buzz_messages_send", + "status": "pending", + "rawInput": { "content": "hi" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-op-ok", + "status": "completed", + "rawOutput": { "stdout": "{\"accepted\":true,\"event_id\":\"e-op\"}" } + }))); + assert!(client.turn_sent_message); + assert!( + client.once_publish_stop_pending, + "first confirmed publish must arm once-publish stop" + ); + + // Second confirmed publish does not re-arm (still pending or already fired). + client.once_publish_stop_pending = false; // simulate fire consumed pending + client.once_publish_stop_fired = true; + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call", + "toolCallId": "tc-op-2", + "title": "buzz_publish__buzz_messages_send", + "status": "pending", + "rawInput": { "content": "again" } + }))); + let _ = client.handle_session_update(&session_update_msg(serde_json::json!({ + "sessionUpdate": "tool_call_update", + "toolCallId": "tc-op-2", + "status": "completed", + "rawOutput": { "stdout": "{\"accepted\":true}" } + }))); + assert!( + !client.once_publish_stop_pending, + "second confirm must not re-arm pending (first already counted)" + ); + } + + /// Intentional once-publish cancel rewrites Cancelled → EndTurn. + #[test] + fn once_publish_stop_rewrites_cancelled_to_end_turn() { + // Build a minimal client via Default-less path: parse_stop_reason needs + // only the once_publish_stop_applied flag. Use spawn_inert in async — + // keep pure unit with a local struct pattern by calling through a + // synthetic result after setting the flag on a real client. + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(async { + let mut client = spawn_inert_client().await; + client.once_publish_stop_applied = true; + let result = serde_json::json!({ "stopReason": "cancelled" }); + let reason = client.parse_stop_reason(&result).unwrap(); + assert_eq!( + reason, + StopReason::EndTurn, + "once-publish cancel must map to EndTurn (no batch requeue)" + ); + assert!( + !client.once_publish_stop_applied, + "applied flag clears after rewrite" + ); + // Plain cancelled without once-publish stays Cancelled. + let reason2 = client + .parse_stop_reason(&serde_json::json!({ "stopReason": "cancelled" })) + .unwrap(); + assert_eq!(reason2, StopReason::Cancelled); + }); + } + /// Argv-style publish (Python subprocess) is recognized as a candidate and /// confirmed on success — the normalization regression from review. #[tokio::test]