fix(acp): coordinate HTTP rate-limit retries

crates/buzz-acp/Cargo.toml

@@ -79,3 +79,4 @@ nix = { version = "0.31", default-features = false, features = ["signal"] }
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }
 httparse = "1"
+wiremock = "0.6"

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_CONTEXT_MESSAGE_LIMIT` | no | `12` | Maximum channel/thread/DM messages the harness supplies per turn (`0` disables; max `100`). |
 | `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). |
@@ -119,6 +120,20 @@ All configuration is via environment variables (or CLI flags — every env var h
 
 **Legacy env vars:** `BUZZ_ACP_PRIVATE_KEY`, `BUZZ_ACP_API_TOKEN`, and `BUZZ_ACP_TURN_TIMEOUT` (replaced by `BUZZ_ACP_IDLE_TIMEOUT`) are still accepted as fallbacks.
 
+### Secure Messaging and Conversation Context
+
+When `BUZZ_ACP_MCP_COMMAND` resolves to `buzz-message-mcp`, the harness passes
+`BUZZ_PRIVATE_KEY` only to that restricted, send-only MCP child. The general
+agent subprocess and its terminal tools do not receive the signing key.
+
+The harness uses its authenticated relay client to supply bounded context before
+each routed turn: recent top-level messages for plain channels, the reply chain
+for threads, and recent conversation messages for DMs. Channel context excludes
+thread replies, is ordered oldest-to-newest, and is capped by
+`BUZZ_ACP_CONTEXT_MESSAGE_LIMIT`. Secure turns must reply through
+`mcp__buzz_message_mcp__send_message`; they are never instructed to fetch
+history or send replies through the terminal CLI.
+
 ### Parallel Agents & Heartbeat
 
 | Flag | Env Var | Default | Description |

crates/buzz-acp/src/acp.rs

@@ -211,6 +211,35 @@ 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,
+    /// Final assistant text streamed during the current prompt. The harness
+    /// uses this as a delivery fallback when an adapter returns text without
+    /// calling the credential-isolated Buzz messaging tool.
+    turn_agent_message: String,
+    /// Whether the current prompt called the credential-isolated sender.
+    turn_secure_message_called: bool,
+    /// Secure sender calls awaiting a terminal ACP status update.
+    turn_secure_message_tool_ids: std::collections::HashSet<String>,
+    /// Tool output may be followed by a replacement final answer. Clear any
+    /// pre-tool narration when the next assistant chunk arrives.
+    turn_reset_message_on_next_chunk: bool,
+}
+
+fn is_secure_message_tool_title(title: &str) -> bool {
+    title == "mcp__buzz_message_mcp__send_message"
+        || (title.contains("buzz-message-mcp") && title.ends_with("send_message"))
+}
+
+fn secure_message_tool_completed_successfully(update: &serde_json::Value) -> bool {
+    update.get("status").and_then(|value| value.as_str()) == Some("completed")
+        && update
+            .get("rawOutput")
+            .and_then(|value| value.get("isError"))
+            .and_then(|value| value.as_bool())
+            != Some(true)
+}
+
+fn record_secure_message_delivery(delivered: &mut bool, update: &serde_json::Value) {
+    *delivered |= secure_message_tool_completed_successfully(update);
 }
 
 /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape
@@ -513,6 +542,13 @@ impl AcpClient {
             cmd.env("CODEX_CONFIG", merged);
         }
 
+        // Signing authority belongs to the harness and the restricted MCP
+        // descriptor, never the general-purpose agent process. `env_remove`
+        // also wins over inherited and persona-provided values.
+        for key in ["BUZZ_PRIVATE_KEY", "BUZZ_AUTH_TAG", "NOSTR_PRIVATE_KEY"] {
+            cmd.env_remove(key);
+        }
+
         // Spawn the agent in its own process group so SIGKILL doesn't propagate
         // to the harness's own process group on Unix.
         // tokio::process::Command::process_group is a stable tokio API (no extra imports needed).
@@ -550,6 +586,10 @@ impl AcpClient {
             steering_supported: false,
             steer_rx: None,
             goose_usage: UsageTracker::default(),
+            turn_agent_message: String::new(),
+            turn_secure_message_called: false,
+            turn_secure_message_tool_ids: std::collections::HashSet::new(),
+            turn_reset_message_on_next_chunk: false,
         })
     }
 
@@ -574,6 +614,14 @@ impl AcpClient {
         self.observer_agent_index
     }
 
+    /// Consume delivery state for a completed turn.
+    pub(crate) fn take_turn_delivery(&mut self) -> (String, bool) {
+        (
+            std::mem::take(&mut self.turn_agent_message),
+            std::mem::take(&mut self.turn_secure_message_called),
+        )
+    }
+
     /// Emit a semantic event to the local observer feed, if enabled.
     pub fn observe(&self, kind: impl Into<String>, payload: serde_json::Value) {
         if let Some(observer) = &self.observer {
@@ -776,6 +824,10 @@ impl AcpClient {
         // prompt so that any setup notifications recorded earlier are not
         // misattributed to this turn.
         self.goose_usage.begin_turn(session_id);
+        self.turn_agent_message.clear();
+        self.turn_secure_message_called = false;
+        self.turn_secure_message_tool_ids.clear();
+        self.turn_reset_message_on_next_chunk = false;
 
         self.last_prompt_id = Some(self.next_id);
         let id = self.next_id;
@@ -1745,6 +1797,11 @@ impl AcpClient {
         match update_type {
             "agent_message_chunk" => {
                 if let Some(text) = update["content"]["text"].as_str() {
+                    if self.turn_reset_message_on_next_chunk {
+                        self.turn_agent_message.clear();
+                        self.turn_reset_message_on_next_chunk = false;
+                    }
+                    self.turn_agent_message.push_str(text);
                     tracing::info!(target: "acp::stream", "{text}");
                 }
                 false
@@ -1758,6 +1815,13 @@ impl AcpClient {
                     .get("kind")
                     .and_then(|v| v.as_str())
                     .unwrap_or("unknown");
+                self.turn_reset_message_on_next_chunk = true;
+                if is_secure_message_tool_title(title) {
+                    if let Some(tool_id) = update.get("toolCallId").and_then(|v| v.as_str()) {
+                        self.turn_secure_message_tool_ids
+                            .insert(tool_id.to_string());
+                    }
+                }
                 tracing::info!(target: "acp::tool", "tool_call: {title} ({kind})");
                 true
             }
@@ -1767,6 +1831,21 @@ impl AcpClient {
                     .and_then(|v| v.as_str())
                     .unwrap_or("?");
                 let status = update.get("status").and_then(|v| v.as_str()).unwrap_or("?");
+                if self.turn_secure_message_tool_ids.contains(tool_id) {
+                    match status {
+                        "completed" => {
+                            record_secure_message_delivery(
+                                &mut self.turn_secure_message_called,
+                                update,
+                            );
+                            self.turn_secure_message_tool_ids.remove(tool_id);
+                        }
+                        "failed" => {
+                            self.turn_secure_message_tool_ids.remove(tool_id);
+                        }
+                        _ => {}
+                    }
+                }
                 tracing::info!(target: "acp::tool", "tool_call_update: {tool_id} → {status}");
                 false
             }
@@ -2893,6 +2972,56 @@ mod tests {
         );
     }
 
+    #[test]
+    fn secure_message_tool_title_is_exactly_scoped() {
+        assert!(is_secure_message_tool_title(
+            "mcp__buzz_message_mcp__send_message"
+        ));
+        assert!(is_secure_message_tool_title(
+            "buzz-message-mcp__send_message"
+        ));
+        assert!(!is_secure_message_tool_title("send_message"));
+        assert!(!is_secure_message_tool_title(
+            "mcp__unrelated__send_message"
+        ));
+    }
+
+    #[test]
+    fn secure_message_completion_rejects_mcp_application_errors() {
+        assert!(secure_message_tool_completed_successfully(
+            &serde_json::json!({"status": "completed"})
+        ));
+        assert!(secure_message_tool_completed_successfully(
+            &serde_json::json!({
+                "status": "completed",
+                "rawOutput": {"isError": false}
+            })
+        ));
+        assert!(!secure_message_tool_completed_successfully(
+            &serde_json::json!({
+                "status": "completed",
+                "rawOutput": {"isError": true}
+            })
+        ));
+        assert!(!secure_message_tool_completed_successfully(
+            &serde_json::json!({"status": "failed"})
+        ));
+    }
+
+    #[test]
+    fn secure_message_delivery_is_monotonic_across_attempts() {
+        let mut delivered = false;
+        record_secure_message_delivery(
+            &mut delivered,
+            &serde_json::json!({"status": "completed", "rawOutput": {"isError": false}}),
+        );
+        record_secure_message_delivery(
+            &mut delivered,
+            &serde_json::json!({"status": "completed", "rawOutput": {"isError": true}}),
+        );
+        assert!(delivered);
+    }
+
     async fn spawn_script(script: &str) -> AcpClient {
         AcpClient::spawn("bash", &["-c".into(), script.into()], &[], false)
             .await
@@ -2971,6 +3100,20 @@ mod tests {
         );
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn spawn_strips_signing_credentials_from_agent_child() {
+        assert_eq!(
+            spawn_named_and_read_child_env(
+                "other-agent",
+                "BUZZ_PRIVATE_KEY",
+                &[("BUZZ_PRIVATE_KEY".into(), "must-not-leak".into())],
+            )
+            .await,
+            "<unset>"
+        );
+    }
+
     #[tokio::test]
     async fn idle_timeout_fires_on_silent_process() {
         let mut client = spawn_script("sleep 10").await;