feat(acp): stream an agent's reply into the channel as it is written

crates/buzz-acp/src/acp.rs

@@ -172,6 +172,10 @@ pub struct AcpClient {
     observer_agent_index: Option<usize>,
     /// Best-effort context attached to raw ACP wire events.
     observer_context: ObserverContext,
+    /// Per-turn sink for `agent_message_chunk` text. Set by the turn driver when
+    /// streaming is enabled; `None` means chunks are logged and dropped, which
+    /// is the behaviour every deployment had before streaming existed.
+    stream_sink: Option<tokio::sync::mpsc::UnboundedSender<String>>,
     /// Most recently observed `_meta.goose.activeRunId` from a
     /// `session/update` notification of kind `session_info_update`.
     ///
@@ -493,13 +497,23 @@ impl AcpClient {
             observer: None,
             observer_agent_index: None,
             observer_context: ObserverContext::default(),
+            stream_sink: None,
             active_run_id: None,
             steer_rx: None,
             goose_usage: UsageTracker::default(),
         })
     }
 
     /// Attach a local observer feed to this ACP client.
+    /// Attach (or clear) the per-turn streaming sink. Cleared between turns so a
+    /// stale sender can never receive a later turn's text.
+    pub(crate) fn set_stream_sink(
+        &mut self,
+        sink: Option<tokio::sync::mpsc::UnboundedSender<String>>,
+    ) {
+        self.stream_sink = sink;
+    }
+
     pub fn set_observer(&mut self, observer: Option<ObserverHandle>, agent_index: usize) {
         self.observer = observer;
         self.observer_agent_index = Some(agent_index);
@@ -1543,6 +1557,12 @@ impl AcpClient {
             "agent_message_chunk" => {
                 if let Some(text) = update["content"]["text"].as_str() {
                     tracing::info!(target: "acp::stream", "{text}");
+                    // Forward to the turn's streamer when one is attached. A
+                    // closed receiver is not an error worth failing a turn over
+                    // — the answer still lands via the normal path.
+                    if let Some(sink) = &self.stream_sink {
+                        let _ = sink.send(text.to_string());
+                    }
                 }
                 false
             }

crates/buzz-acp/src/config.rs

@@ -261,6 +261,22 @@ pub struct CliArgs {
     #[arg(long, env = "BUZZ_ACP_MCP_COMMAND", default_value = "")]
     pub mcp_command: String,
 
+    /// Stream the agent's reply into the channel as it is generated: post once,
+    /// then edit that message as more text arrives. Off by default — it changes
+    /// what every channel sees.
+    #[arg(long, env = "BUZZ_ACP_STREAM_REPLIES")]
+    pub stream_replies: bool,
+
+    /// Minimum milliseconds between streamed edits. The first publish of a turn
+    /// ignores this, so the channel sees the agent start work immediately.
+    #[arg(long, env = "BUZZ_ACP_STREAM_THROTTLE_MS", default_value_t = 1_000)]
+    pub stream_throttle_ms: u64,
+
+    /// Minimum newly-buffered characters before a mid-turn edit is worth a round
+    /// trip. Guards against an edit per token.
+    #[arg(long, env = "BUZZ_ACP_STREAM_MIN_DELTA", default_value_t = 24)]
+    pub stream_min_delta: usize,
+
     /// Idle timeout: max seconds of silence before killing a turn.
     /// Resets on any agent stdout activity.
     #[arg(long, env = "BUZZ_ACP_IDLE_TIMEOUT")]
@@ -495,6 +511,7 @@ pub struct Config {
     pub agent_command: String,
     pub agent_args: Vec<String>,
     pub mcp_command: String,
+    pub stream_flush: crate::stream_flush::StreamFlushConfig,
     pub idle_timeout_secs: u64,
     pub max_turn_duration_secs: u64,
     pub agents: u32,
@@ -1035,6 +1052,11 @@ impl Config {
             agent_command,
             agent_args,
             mcp_command: args.mcp_command,
+            stream_flush: crate::stream_flush::StreamFlushConfig {
+                enabled: args.stream_replies,
+                throttle_ms: args.stream_throttle_ms,
+                min_delta_chars: args.stream_min_delta,
+            },
             idle_timeout_secs,
             max_turn_duration_secs,
             agents: args.agents,
@@ -1413,6 +1435,7 @@ mod tests {
             agent_command: "goose".into(),
             agent_args: vec!["acp".into()],
             mcp_command: "".into(),
+            stream_flush: crate::stream_flush::StreamFlushConfig::default(),
             idle_timeout_secs: DEFAULT_IDLE_TIMEOUT_SECS,
             max_turn_duration_secs: DEFAULT_MAX_TURN_DURATION_SECS,
             agents: 1,

crates/buzz-acp/src/lib.rs

@@ -10,6 +10,7 @@ mod pool_lifecycle;
 mod queue;
 mod relay;
 mod setup_mode;
+pub(crate) mod stream_flush;
 mod usage;
 
 pub use usage::TurnUsage;
@@ -1528,6 +1529,7 @@ async fn tokio_main() -> Result<()> {
 
     let base_prompt_content = config.base_prompt_content.take();
     let ctx = Arc::new(PromptContext {
+        stream_flush: config.stream_flush,
         mcp_servers: build_mcp_servers(&config),
         initial_message: config.initial_message.clone(),
         idle_timeout: Duration::from_secs(config.idle_timeout_secs),
@@ -4963,6 +4965,7 @@ mod build_mcp_servers_tests {
             agent_command: "goose".into(),
             agent_args: vec!["acp".into()],
             mcp_command: "test-mcp-server".into(),
+            stream_flush: stream_flush::StreamFlushConfig::default(),
             idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS,
             max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS,
             agents: 1,
@@ -5184,6 +5187,7 @@ mod error_outcome_emission_tests {
             agent_command: "true".into(),
             agent_args: vec![],
             mcp_command: "test-mcp-server".into(),
+            stream_flush: stream_flush::StreamFlushConfig::default(),
             idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS,
             max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS,
             agents: 1,

crates/buzz-acp/src/pool.rs

@@ -480,6 +480,8 @@ impl ChannelInfoResolver {
 }
 
 pub struct PromptContext {
+    /// Streaming policy for replies. Disabled by default.
+    pub stream_flush: crate::stream_flush::StreamFlushConfig,
     pub mcp_servers: Vec<McpServer>,
     pub initial_message: Option<String>,
     pub idle_timeout: Duration,
@@ -1879,6 +1881,41 @@ pub async fn run_prompt_task(
         None => prompt_sections.iter().map(String::as_str).collect(),
     };
 
+    // Streamed replies: attach a chunk sink for the duration of the prompt so
+    // the channel sees the answer being written instead of arriving whole.
+    //
+    // `stream_tx` is scope-local ON PURPOSE. Dropping it is the end-of-turn
+    // signal, so every exit below — success, cancel, idle timeout, agent exit,
+    // early return — flushes the tail without needing to remember to. Only
+    // channel turns stream; a heartbeat has no channel to write into.
+    let stream_handle = match (ctx.stream_flush.enabled, &source) {
+        (true, PromptSource::Channel(channel_id)) => {
+            let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+            let publisher = RelayStreamPublisher {
+                rest: ctx.rest_client.clone(),
+                channel_id: *channel_id,
+                thread_tags: batch
+                    .as_ref()
+                    .and_then(|b| b.events.last())
+                    .map(|be| crate::queue::parse_thread_tags(&be.event))
+                    .unwrap_or_default(),
+            };
+            let cfg = ctx.stream_flush;
+            agent.acp.set_stream_sink(Some(tx.clone()));
+            let join = tokio::spawn(async move {
+                crate::stream_flush::run_stream(rx, publisher, cfg, || {
+                    std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .map(|d| d.as_millis() as u64)
+                        .unwrap_or(0)
+                })
+                .await
+            });
+            Some((tx, join))
+        }
+        _ => None,
+    };
+
     // 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.
@@ -2046,6 +2083,19 @@ pub async fn run_prompt_task(
         }
     };
 
+    // Turn is over: detach the sink and drop the sender so the streamer sees the
+    // channel close, performs its final flush, and exits. Awaiting the join
+    // means the complete reply is on the relay before the outcome is reported —
+    // otherwise a caller could observe "turn complete" against a message still
+    // missing its last edit.
+    if let Some((tx, join)) = stream_handle {
+        agent.acp.set_stream_sink(None);
+        drop(tx);
+        if let Err(e) = join.await {
+            tracing::debug!("stream task join failed: {e}");
+        }
+    }
+
     match prompt_result {
         Ok(stop_reason) => {
             log_stop_reason(&source, &stop_reason);
@@ -3706,6 +3756,69 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec<String>)
     }
 }
 
+/// Publishes a streamed reply into a channel: one kind:9 post, then kind:40003
+/// edits against it.
+///
+/// Edits are deliberately unthreaded — they carry only the `e` tag identifying
+/// the message being replaced, so a stream never fans out into the thread.
+struct RelayStreamPublisher {
+    rest: crate::relay::RestClient,
+    channel_id: Uuid,
+    thread_tags: ThreadTags,
+}
+
+impl crate::stream_flush::StreamPublisher for RelayStreamPublisher {
+    async fn post(&mut self, body: String) -> Option<String> {
+        let thread_ref = self.thread_tags.root_event_id.as_deref().and_then(|root| {
+            let root_id = nostr::EventId::from_hex(root).ok()?;
+            let parent_id = self
+                .thread_tags
+                .parent_event_id
+                .as_deref()
+                .and_then(|p| nostr::EventId::from_hex(p).ok())
+                .unwrap_or(root_id);
+            Some(buzz_sdk::ThreadRef {
+                root_event_id: root_id,
+                parent_event_id: parent_id,
+            })
+        });
+        let builder =
+            buzz_sdk::build_message(self.channel_id, &body, thread_ref.as_ref(), &[], false, &[])
+                .ok()?;
+        let event = builder.sign_with_keys(&self.rest.keys).ok()?;
+        match tokio::time::timeout(Duration::from_secs(5), self.rest.submit_event(&event)).await {
+            Ok(Ok(_)) => Some(event.id.to_hex()),
+            Ok(Err(e)) => {
+                tracing::debug!(channel = %self.channel_id, "stream post failed: {e}");
+                None
+            }
+            Err(_) => {
+                tracing::debug!(channel = %self.channel_id, "stream post timed out");
+                None
+            }
+        }
+    }
+
+    async fn edit(&mut self, event_id: &str, body: String) {
+        let Ok(target) = nostr::EventId::from_hex(event_id) else {
+            return;
+        };
+        let Ok(builder) = buzz_sdk::build_edit(self.channel_id, target, &body) else {
+            return;
+        };
+        let Ok(event) = builder.sign_with_keys(&self.rest.keys) else {
+            return;
+        };
+        // Best effort: a dropped edit only costs freshness — the next edit, or
+        // the end-of-turn flush, carries the full text anyway.
+        match tokio::time::timeout(Duration::from_secs(5), self.rest.submit_event(&event)).await {
+            Ok(Ok(_)) => {}
+            Ok(Err(e)) => tracing::debug!(channel = %self.channel_id, "stream edit failed: {e}"),
+            Err(_) => tracing::debug!(channel = %self.channel_id, "stream edit timed out"),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -5335,6 +5448,7 @@ mod tests {
     ) -> PromptContext {
         use crate::relay::RestClient;
         PromptContext {
+            stream_flush: crate::stream_flush::StreamFlushConfig::default(),
             mcp_servers: vec![],
             initial_message: None,
             idle_timeout: Duration::from_secs(60),