From 9aca746c1fe6d70d357b207232073b951bb8cf30 Mon Sep 17 00:00:00 2001 From: johncarle Date: Thu, 13 Aug 2026 11:13:11 -0600 Subject: [PATCH] Publish successful ACP replies Retain streamed assistant text and the triggering batch, then sign and publish a threaded kind-9 reply for successful managed-agent turns. Add coverage for chunk accumulation, threading, mentions, signing, and empty responses. Co-authored-by: johncarle Signed-off-by: johncarle --- crates/buzz-acp/src/acp.rs | 29 +++++++ crates/buzz-acp/src/lib.rs | 30 +++++++ crates/buzz-acp/src/pool.rs | 156 +++++++++++++++++++++++++++++++++++- 3 files changed, 213 insertions(+), 2 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index f04b8eeec0..2fc18d59cd 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -214,6 +214,8 @@ pub struct AcpClient { standard_usage: StandardUsageTracker, /// Known adapter identity for prompt-response usage mapping. standard_adapter: Option, + /// Text emitted by `agent_message_chunk` updates for the active prompt. + current_agent_response: String, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -563,6 +565,7 @@ impl AcpClient { goose_usage: UsageTracker::default(), standard_usage: StandardUsageTracker::default(), standard_adapter, + current_agent_response: String::new(), }) } @@ -599,6 +602,11 @@ impl AcpClient { } } + /// Take the assistant text accumulated during the most recent prompt. + pub(crate) fn take_agent_response_text(&mut self) -> String { + std::mem::take(&mut self.current_agent_response) + } + /// Send the `initialize` request and return the agent's response result value. /// /// Must be called exactly once, before any other ACP method. @@ -781,6 +789,7 @@ impl AcpClient { idle_timeout: std::time::Duration, max_duration: std::time::Duration, ) -> Result { + self.current_agent_response.clear(); let params = build_prompt_params(session_id, prompt_blocks); let hard_deadline = tokio::time::Instant::now() + max_duration; self.current_hard_deadline = Some(hard_deadline); @@ -1755,6 +1764,7 @@ impl AcpClient { match update_type { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { + self.current_agent_response.push_str(text); tracing::info!(target: "acp::stream", "{text}"); } false @@ -2959,6 +2969,25 @@ mod tests { .expect("failed to spawn test script") } + #[tokio::test] + async fn agent_message_chunks_accumulate_and_take_clears_response() { + let mut client = spawn_script("sleep 10").await; + for text in ["hello", " ", "world"] { + let update = serde_json::json!({ + "params": { + "update": { + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": text } + } + } + }); + let _ = client.handle_session_update(&update); + } + + assert_eq!(client.take_agent_response_text(), "hello world"); + assert_eq!(client.take_agent_response_text(), ""); + } + #[cfg(unix)] async fn spawn_named_script(name: &str, script: &str) -> (AcpClient, std::path::PathBuf) { use std::os::unix::fs::PermissionsExt; diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 27b9000b7b..0569c6a49f 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3677,6 +3677,20 @@ fn spawn_failure_notice( } } +/// Spawn a task that publishes the visible text from a successful ACP turn. +fn spawn_success_reply( + rest_client: Option<&relay::RestClient>, + batch: FlushBatch, + content: String, +) { + if let Some(rest) = rest_client { + let rest = rest.clone(); + tokio::spawn(async move { + pool::post_success_reply(&rest, &batch, &content).await; + }); + } +} + #[allow(clippy::too_many_arguments)] fn handle_prompt_result( pool: &mut AgentPool, @@ -3718,6 +3732,16 @@ fn handle_prompt_result( } } + // Successful turns retain the triggering batch solely so the streamed ACP + // text can be published with the correct channel, thread, and recipient. + // Remove it before the failure/requeue state machine below, where any + // remaining batch necessarily belongs to an unsuccessful turn. + let success_batch = if matches!(result.outcome, PromptOutcome::Ok(_)) { + result.batch.take() + } else { + None + }; + // The hard-timeout death_message (below) must describe the batch's // *actual* fate, not just the `recently_active` eligibility flag — a // recently-active batch that exhausts the retry budget in queue.requeue() @@ -3897,6 +3921,12 @@ fn handle_prompt_result( match result.outcome { // Successful prompt — return agent to pool. PromptOutcome::Ok(_) => { + let response_text = result.agent.acp.take_agent_response_text(); + if let Some(batch) = success_batch { + if !removed_channels.contains(&batch.channel_id) { + spawn_success_reply(rest_client, batch, response_text); + } + } tracing::debug!( agent = agent_index, outcome = outcome_label, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 2efacce2b1..a12c6bde3f 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2350,7 +2350,7 @@ pub async fn run_prompt_task( agent, source, PromptOutcome::Ok(StopReason::EndTurn), - None, // turn succeeded — batch was processed, no requeue + batch, ); return; } @@ -2424,7 +2424,7 @@ pub async fn run_prompt_task( agent, source, PromptOutcome::Ok(stop_reason), - None, + batch, ); } Err(AcpError::AgentExited) => { @@ -4281,6 +4281,78 @@ pub(crate) async fn post_failure_notice( } } +/// Best-effort: publish the assistant text produced by a successful ACP turn. +/// +/// ACP streams visible text through `agent_message_chunk` notifications and +/// returns only a stop reason from `session/prompt`. The managed harness owns +/// the signing key, so it must publish the accumulated text on the agent's +/// behalf; the sandboxed agent process cannot sign this event itself. +pub(crate) async fn post_success_reply( + rest: &crate::relay::RestClient, + batch: &FlushBatch, + content: &str, +) { + let Some(event) = build_success_reply_event(&rest.keys, batch, content) else { + return; + }; + match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await { + Ok(Ok(_)) => {} + Ok(Err(e)) => tracing::warn!(channel = %batch.channel_id, "successful reply failed: {e}"), + Err(_) => tracing::warn!(channel = %batch.channel_id, "successful reply timed out"), + } +} + +fn build_success_reply_event( + keys: &nostr::Keys, + batch: &FlushBatch, + content: &str, +) -> Option { + let trigger = batch.events.last().map(|event| &event.event)?; + if content.trim().is_empty() { + tracing::debug!(channel = %batch.channel_id, "successful turn produced no visible text"); + return None; + } + + // Keep replies flat within an existing thread. For a top-level trigger, + // the trigger itself becomes both root and parent. + let thread_tags = crate::queue::parse_thread_tags(trigger); + let thread_ref = if let Some(root) = thread_tags.root_event_id.as_deref() { + nostr::EventId::from_hex(root) + .ok() + .map(|root_event_id| buzz_sdk::ThreadRef { + root_event_id, + parent_event_id: root_event_id, + }) + } else { + Some(buzz_sdk::ThreadRef { + root_event_id: trigger.id, + parent_event_id: trigger.id, + }) + }; + let author = trigger.pubkey.to_hex(); + let builder = match buzz_sdk::build_message( + batch.channel_id, + content, + thread_ref.as_ref(), + &[&author], + false, + &[], + ) { + Ok(builder) => builder, + Err(e) => { + tracing::warn!(channel = %batch.channel_id, "successful reply: build failed: {e}"); + return None; + } + }; + match builder.sign_with_keys(keys) { + Ok(event) => Some(event), + Err(e) => { + tracing::warn!(channel = %batch.channel_id, "successful reply: sign failed: {e}"); + None + } + } +} + /// 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 @@ -4402,6 +4474,86 @@ mod tests { use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; + fn success_reply_batch(trigger: nostr::Event, channel_id: Uuid) -> FlushBatch { + FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event: trigger, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + } + } + + #[test] + fn successful_reply_is_signed_threaded_and_mentions_trigger_author() { + let channel_id = Uuid::new_v4(); + let author_keys = Keys::generate(); + let agent_keys = Keys::generate(); + let root = EventBuilder::new(Kind::Custom(9), "root") + .sign_with_keys(&author_keys) + .unwrap(); + let trigger = EventBuilder::new(Kind::Custom(9), "question") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["e", &root.id.to_hex(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&author_keys) + .unwrap(); + let event = build_success_reply_event( + &agent_keys, + &success_reply_batch(trigger.clone(), channel_id), + "answer", + ) + .expect("visible reply event"); + + assert_eq!(event.kind, Kind::Custom(9)); + assert_eq!(event.content, "answer"); + assert_eq!(event.pubkey, agent_keys.public_key()); + assert!(event.verify().is_ok()); + let tags: Vec<&[String]> = event.tags.iter().map(|tag| tag.as_slice()).collect(); + assert!(tags + .iter() + .any(|tag| tag.len() >= 2 && tag[0] == "h" && tag[1] == channel_id.to_string())); + assert!(tags.iter().any(|tag| tag.len() >= 4 + && tag[0] == "e" + && tag[1] == root.id.to_hex() + && tag[3] == "root")); + assert!(tags.iter().any(|tag| tag.len() >= 4 + && tag[0] == "e" + && tag[1] == root.id.to_hex() + && tag[3] == "reply")); + assert!(tags + .iter() + .any(|tag| tag.len() >= 2 && tag[0] == "p" && tag[1] == trigger.pubkey.to_hex())); + } + + #[test] + fn successful_reply_uses_top_level_trigger_as_root_and_skips_blank_text() { + let channel_id = Uuid::new_v4(); + let author_keys = Keys::generate(); + let agent_keys = Keys::generate(); + let trigger = EventBuilder::new(Kind::Custom(9), "question") + .sign_with_keys(&author_keys) + .unwrap(); + let batch = success_reply_batch(trigger.clone(), channel_id); + + assert!(build_success_reply_event(&agent_keys, &batch, " \n\t").is_none()); + let event = + build_success_reply_event(&agent_keys, &batch, "answer").expect("visible reply event"); + let tags: Vec<&[String]> = event.tags.iter().map(|tag| tag.as_slice()).collect(); + assert!(tags.iter().any(|tag| tag.len() >= 4 + && tag[0] == "e" + && tag[1] == trigger.id.to_hex() + && tag[3] == "root")); + assert!(tags.iter().any(|tag| tag.len() >= 4 + && tag[0] == "e" + && tag[1] == trigger.id.to_hex() + && tag[3] == "reply")); + } + fn test_mcp_server() -> McpServer { McpServer { name: "dev".into(),