fix(acp): confirm message delivery from tool outcome, not input text

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 <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
(cherry picked from commit e992ae745d)
This commit is contained in:
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
2026-07-24 15:19:02 +10:00
committed by Michael Neale
parent 43a640b2e0
commit 977457865b
+430 -27
View File
@@ -207,10 +207,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<String>,
}
/// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape
@@ -509,6 +519,7 @@ impl AcpClient {
goose_usage: UsageTracker::default(),
turn_message_text: String::new(),
turn_sent_message: false,
turn_publish_candidates: std::collections::HashSet::new(),
})
}
@@ -705,6 +716,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;
@@ -802,14 +814,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<String> {
let text = std::mem::take(&mut self.turn_message_text);
let sent = self.turn_sent_message;
@@ -1592,16 +1605,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
}
@@ -1612,6 +1636,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" => {
@@ -2003,12 +2045,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
@@ -2021,11 +2070,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::<Vec<_>>().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<bool> {
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<i64> {
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
@@ -2153,6 +2303,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",
@@ -2168,6 +2329,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.
@@ -2194,6 +2437,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));