diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index e6164b02dd3..e779e133f7b 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -111,6 +111,7 @@ All configuration is via environment variables (or CLI flags — every env var h | `BUZZ_ACP_AGENT_COMMAND` | no | `goose` | Agent binary to spawn. | | `BUZZ_ACP_AGENT_ARGS` | no | `acp` | Agent arguments (comma-separated). | | `BUZZ_ACP_MCP_COMMAND` | no | `""` (empty) | Path to an optional MCP server binary to provide to the agent subprocess. | +| `BUZZ_ACP_PUBLISH_AGENT_OUTPUT` | no | `false` | Publish the ACP agent's generated text as a threaded Buzz reply after each successful channel turn. Use this for discussion-only agents that do not have a messaging MCP tool. | | `BUZZ_ACP_IDLE_TIMEOUT` | no | `620` | Idle timeout: max seconds of silence before cancelling a turn. Resets on any agent stdout activity. | | `BUZZ_ACP_MAX_TURN_DURATION` | no | `7200` | Absolute wall-clock cap per turn (safety valve). | | `BUZZ_API_TOKEN` | no | — | API token (required if relay enforces token auth). | diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 700d5e8dcfd..c107ecaa6f7 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -211,6 +211,9 @@ 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, + /// Text emitted through `agent_message_chunk` notifications during the + /// current turn. Drained by the harness when auto-publish is enabled. + agent_output: String, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -550,6 +553,7 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), + agent_output: String::new(), }) } @@ -768,6 +772,7 @@ impl AcpClient { idle_timeout: std::time::Duration, max_duration: std::time::Duration, ) -> Result { + self.agent_output.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); @@ -881,6 +886,11 @@ impl AcpClient { self.goose_usage.take() } + /// Drain text emitted by the agent during the most recent prompt turn. + pub fn take_agent_output(&mut self) -> String { + std::mem::take(&mut self.agent_output) + } + /// Install a per-turn steer request channel for goose-native /// non-cancelling mid-turn delivery. /// @@ -1732,6 +1742,7 @@ impl AcpClient { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { tracing::info!(target: "acp::stream", "{text}"); + self.agent_output.push_str(text); } false } @@ -3554,6 +3565,31 @@ mod tests { .expect("spawn cat as inert client") } + #[tokio::test] + async fn agent_message_chunks_are_collected_for_turn_output() { + let mut client = spawn_inert_client().await; + for text in ["Hello", " from", " Goose"] { + let msg = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": text}, + }, + } + }); + let _ = client.handle_session_update(&msg); + } + + assert_eq!(client.take_agent_output(), "Hello from Goose"); + assert!( + client.take_agent_output().is_empty(), + "taking drains output" + ); + } + /// Build a `session/update` JSON-RPC notification carrying a /// `session_info_update` with the given `_meta.goose.activeRunId` value. /// Pass `None` to omit the `activeRunId` field entirely. diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 35aaec188db..65210cade32 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -409,6 +409,12 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_NO_BASE_PROMPT")] pub no_base_prompt: bool, + /// Publish the ACP agent's generated text as a Buzz reply after each + /// successful channel turn. Disabled by default to preserve tool-driven + /// agents that publish their own messages. + #[arg(long, env = "BUZZ_ACP_PUBLISH_AGENT_OUTPUT", default_value_t = false)] + pub publish_agent_output: bool, + /// Path to a custom base prompt file. Overrides the compiled-in default. /// Mutually exclusive with --no-base-prompt. #[arg( @@ -564,6 +570,8 @@ pub struct Config { pub agent_owner: Option, /// Disable the [Base] platform-context section prepended to every prompt. pub no_base_prompt: bool, + /// Publish ACP `agent_message_chunk` text as a Buzz reply on success. + pub publish_agent_output: bool, /// Resolved content from `--base-prompt-file`, read and validated in /// `from_cli()`. `None` when using the compiled-in default or when /// `--no-base-prompt` is set. @@ -1109,6 +1117,7 @@ impl Config { lazy_pool: args.lazy_pool, agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()), no_base_prompt: args.no_base_prompt, + publish_agent_output: args.publish_agent_output, base_prompt_content, }; @@ -1480,6 +1489,7 @@ mod tests { lazy_pool: false, agent_owner: None, no_base_prompt: false, + publish_agent_output: false, base_prompt_content: None, } } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index b69a1f453a5..fefe97f2634 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1843,6 +1843,8 @@ async fn tokio_main() -> Result<()> { memory_enabled: config.memory_enabled, harness_name: crate::config::normalize_agent_command_identity(&config.agent_command), relay_url: config.relay_url.clone(), + event_publisher: Some(presence_publisher.clone()), + publish_agent_output: config.publish_agent_output, }); if !config.memory_enabled { @@ -6210,6 +6212,7 @@ mod build_mcp_servers_tests { lazy_pool: false, agent_owner: None, no_base_prompt: false, + publish_agent_output: false, base_prompt_content: None, } } @@ -6432,6 +6435,7 @@ mod error_outcome_emission_tests { lazy_pool: false, agent_owner: None, no_base_prompt: false, + publish_agent_output: false, base_prompt_content: None, } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index ddc0330d9f2..488b3f416f0 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -40,7 +40,7 @@ use crate::queue::{ CancelReason, ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, PromptProfile, PromptProfileLookup, ThreadTags, }; -use crate::relay::{ChannelInfo, RestClient}; +use crate::relay::{ChannelInfo, RelayEventPublisher, RestClient}; /// Window within which agent activity before a hard-cap death qualifies /// the turn as "recently active" (eligible for requeue instead of dead-letter). @@ -564,6 +564,64 @@ pub struct PromptContext { /// the desktop keys per (agent, relay) pair, e.g. `session_config_captured`, /// mirroring the `managed_agent_runtime_lifecycle` frames. pub relay_url: String, + /// Relay publisher used by the optional ACP-output bridge. + pub event_publisher: Option, + /// Publish collected ACP agent text as a Buzz reply after successful turns. + pub publish_agent_output: bool, +} + +async fn publish_agent_output( + ctx: &PromptContext, + batch: &FlushBatch, + content: &str, +) -> anyhow::Result<()> { + let triggering_event = &batch + .events + .last() + .ok_or_else(|| anyhow::anyhow!("cannot publish agent output for empty batch"))? + .event; + let event = + build_agent_output_event(&ctx.agent_keys, batch.channel_id, triggering_event, content)?; + ctx.event_publisher + .as_ref() + .ok_or_else(|| anyhow::anyhow!("agent output publisher is not configured"))? + .publish_event(event) + .await + .map_err(|e| anyhow::anyhow!("failed to publish agent output reply: {e}"))?; + Ok(()) +} + +fn build_agent_output_event( + agent_keys: &nostr::Keys, + channel_id: Uuid, + triggering_event: &nostr::Event, + content: &str, +) -> anyhow::Result { + use buzz_sdk::ThreadRef; + + let thread_tags = crate::queue::parse_thread_tags(triggering_event); + let reply_id = match thread_tags.root_event_id { + Some(root) => nostr::EventId::from_hex(&root) + .map_err(|e| anyhow::anyhow!("invalid root event id: {e}"))?, + None => triggering_event.id, + }; + let thread_ref = ThreadRef { + root_event_id: reply_id, + parent_event_id: reply_id, + }; + let author = triggering_event.pubkey.to_hex(); + let builder = buzz_sdk::build_message( + channel_id, + content, + Some(&thread_ref), + &[&author], + false, + &[], + ) + .map_err(|e| anyhow::anyhow!("failed to build agent output reply: {e}"))?; + builder + .sign_with_keys(agent_keys) + .map_err(|e| anyhow::anyhow!("failed to sign agent output reply: {e}")) } impl AgentPool { @@ -2172,6 +2230,21 @@ pub async fn run_prompt_task( } let core_stop = acp_stop_to_core(&stop_reason); + let agent_output = agent.acp.take_agent_output(); + if ctx.publish_agent_output { + if let (PromptSource::Channel(_), Some(batch)) = (&source, batch.as_ref()) { + let content = agent_output.trim(); + if !content.is_empty() { + if let Err(e) = publish_agent_output(&ctx, batch, content).await { + tracing::error!( + agent_index = agent.index, + channel_id = %batch.channel_id, + "{e}" + ); + } + } + } + } let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( &ctx, @@ -6494,6 +6567,44 @@ mod tests { make_prompt_context_impl(&agent_keys, None) } + #[test] + fn agent_output_event_replies_to_triggering_conversation() { + use buzz_sdk::ThreadRef; + + let channel_id = Uuid::new_v4(); + let author_keys = nostr::Keys::generate(); + let agent_keys = nostr::Keys::generate(); + let root = buzz_sdk::build_message(channel_id, "root", None, &[], false, &[]) + .unwrap() + .sign_with_keys(&author_keys) + .unwrap(); + let thread_ref = ThreadRef { + root_event_id: root.id, + parent_event_id: root.id, + }; + let trigger = + buzz_sdk::build_message(channel_id, "question", Some(&thread_ref), &[], false, &[]) + .unwrap() + .sign_with_keys(&author_keys) + .unwrap(); + + let output = build_agent_output_event(&agent_keys, channel_id, &trigger, "answer") + .expect("agent output event"); + let value = serde_json::to_value(output).unwrap(); + let tags = value["tags"].as_array().unwrap(); + + assert_eq!(value["content"], "answer"); + assert!(tags + .iter() + .any(|tag| tag[0] == "h" && tag[1] == channel_id.to_string())); + assert!(tags + .iter() + .any(|tag| tag[0] == "e" && tag[1] == root.id.to_hex())); + assert!(tags + .iter() + .any(|tag| { tag[0] == "p" && tag[1] == author_keys.public_key().to_hex() })); + } + fn make_prompt_context_with_owner( agent_keys: &nostr::Keys, owner_pubkey: nostr::PublicKey, @@ -6542,6 +6653,8 @@ mod tests { memory_enabled: false, harness_name: "goose".to_string(), relay_url: "ws://127.0.0.1:3000".to_string(), + event_publisher: None, + publish_agent_output: false, } }