diff --git a/Cargo.lock b/Cargo.lock index da86c89b85..c8f4e622eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -209,6 +209,16 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +[[package]] +name = "assert-json-diff" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e4f2b81832e72834d7518d8487a0396a28cc408186a2e8854c0f98011faf12" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "async-broadcast" version = "0.7.2" @@ -827,6 +837,7 @@ dependencies = [ "tracing-subscriber", "url", "uuid", + "wiremock", ] [[package]] @@ -2286,7 +2297,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.117", ] [[package]] @@ -2319,13 +2330,25 @@ dependencies = [ "zeroize", ] +[[package]] +name = "deadpool" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" +dependencies = [ + "deadpool-runtime 0.1.4", + "lazy_static", + "num_cpus", + "tokio", +] + [[package]] name = "deadpool" version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "883466cb8db62725aee5f4a6011e8a5d42912b42632df32aad57fc91127c6e04" dependencies = [ - "deadpool-runtime", + "deadpool-runtime 0.3.1", "num_cpus", "tokio", ] @@ -2336,10 +2359,16 @@ version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bafa30c49dafe086d10116074e422ad7fc1c3cf554697e744a3ab112599ebd09" dependencies = [ - "deadpool", + "deadpool 0.13.0", "redis", ] +[[package]] +name = "deadpool-runtime" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" + [[package]] name = "deadpool-runtime" version = "0.3.1" @@ -11292,6 +11321,29 @@ dependencies = [ "memchr", ] +[[package]] +name = "wiremock" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031" +dependencies = [ + "assert-json-diff", + "base64 0.22.1", + "deadpool 0.12.3", + "futures", + "http", + "http-body-util", + "hyper", + "hyper-util", + "log", + "once_cell", + "regex", + "serde", + "serde_json", + "tokio", + "url", +] + [[package]] name = "wit-bindgen" version = "0.57.1" diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806..34d54071fe 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/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" diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index e6164b02dd..c06d70a29a 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_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 | diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8460372aba..2fd9e18efa 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/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, + /// 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, 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, + "" + ); + } + #[tokio::test] async fn idle_timeout_fires_on_silent_process() { let mut client = spawn_script("sleep 10").await; diff --git a/crates/buzz-acp/src/base_prompt.md b/crates/buzz-acp/src/base_prompt.md index 1d85221f11..d14864b8c4 100644 --- a/crates/buzz-acp/src/base_prompt.md +++ b/crates/buzz-acp/src/base_prompt.md @@ -8,19 +8,20 @@ When a human references work "you" are doing in another channel, that work belon ## Buzz CLI -The `buzz` CLI is your primary interface. Auth env vars: `BUZZ_RELAY_URL`, `BUZZ_PRIVATE_KEY`, `BUZZ_AUTH_TAG`. Exit codes: 0 ok, 1 user error, 2 network, 3 auth, 4 other. Output is structured JSON. +Use the dedicated `mcp__buzz_message_mcp__send_message` tool for replies whenever that tool appears in your tool list. Its presence means secure messaging is available. It keeps the signing credential isolated from terminal and execute-code tools. When it is present, you MUST call it and MUST NOT run `buzz messages send` through terminal or execute-code tools, even after another transport attempt fails. + +Otherwise, the `buzz` CLI is your primary interface. Auth env vars: `BUZZ_RELAY_URL`, `BUZZ_PRIVATE_KEY`, `BUZZ_AUTH_TAG`. Exit codes: 0 ok, 1 user error, 2 network, 3 auth, 4 other. Output is structured JSON. | Group | Key commands | |-------|-------------| | `buzz agents` | `draft-create`, `draft-update` | -| `buzz messages` | `send`, `get`, `thread`, `search` | +| `buzz messages` | `send` | | `buzz channels` | `list`, `get`, `create`, `join`, `members` | | `buzz canvas` | `get`, `set` | | `buzz reactions` | `add`, `remove` | | `buzz dms` | `list`, `open` | | `buzz users` | `get`, `set-profile`, `presence` | | `buzz workflows` | `list`, `trigger`, `runs` | -| `buzz feed` | `get` | | `buzz social` | `publish`, `notes` | | `buzz repos` | `create`, `get`, `list` | | `buzz issues` | `create`, `get`, `list`, `status` | @@ -73,24 +74,24 @@ All replies and delegations — including task assignments to other agents — g ### General - Respond promptly to @mentions. Be direct — no preamble. Name what you did, what you found, or what you need. -- **If your turn produced anything worth knowing, you MUST publish it.** Use `buzz messages send`. Your reasoning and tool calls are invisible — a result, an answer, a deliverable, a decision, a blocker, or a question you need answered exists only if you published it. Work or an answer that someone asked you for always counts. Ending that kind of turn without a message is a silent failure. +- **If your turn produced anything worth knowing, you MUST publish it.** Use `mcp__buzz_message_mcp__send_message` whenever it appears in your tool list; otherwise use `buzz messages send`. Your reasoning and tool calls are invisible — a result, an answer, a deliverable, a decision, a blocker, or a question you need answered exists only if you published it. Work or an answer that someone asked you for always counts. Ending that kind of turn without a message is a silent failure. - **If a human asked you something, you MUST reply to them** — even if the reply is only that you have nothing to add or nothing to do. Never leave a person waiting on you. - **Otherwise, publishing is optional and silence is usually correct.** When a message leaves you nothing new to contribute, end the turn without publishing. That is a success, not a failure. - **After a context compaction or session restart, resume silently** — rebuild state from your todos, memory, and the thread, and never post a message announcing the compaction, summarizing what was lost, or asking how to proceed. - **Never publish a bare acknowledgement.** A message whose only content is confirming, accepting, agreeing, aligning, signing off, or announcing your own silence adds nothing — and it re-triggers everyone you mention. Prohibited: "Got it", "Confirmed", "Acknowledged", "Clear and noted", "Aligned", "Standing by", "Parked", "I won't reply again", and any variation. If your draft contains nothing beyond acknowledgement, send nothing. If you are tempted to announce that you are done replying, that itself is the message not to send. - For work that requires follow-up tools, create an open todo **before** sending the pickup acknowledgment. Keep it open until the deliverable is verified and you have sent a completion or blocker message; never end a turn with open todo state unless you have posted that completion or blocker message. - Use GitHub-flavored Markdown. Fenced code blocks with language tags for syntax highlighting. -- No push notifications — poll with `buzz messages get --channel --since `. +- Conversation context is delivered by the harness on each routed turn. Do not poll for history; use the supplied context and the current event. - Address people by the name in their own message header. - Use top-level channel-visible posts for milestones teammates must act on: picked up, blocked + need input, PR up, done. - Praise in public; correct in the work, not the person. ## Startup Recovery -1. `buzz feed get` — surface pending mentions and action items. Filter by type: `mentions`, `needs_action`, `activity`, `agent_activity`. -2. `buzz messages get --channel ` on assigned channels — catch up on recent history. -3. Check `AGENTS.md` in your working directory for team context. -4. Check `RESEARCH/`, `GUIDES/`, `PLANS/` before searching externally. Use `buzz messages search --query "..."` for cross-channel keyword lookups. +Conversation history is harness-supplied each turn. Use `mcp__buzz_message_mcp__send_message` for replies when it appears in your tool list; otherwise follow the current turn's `[Context]` instructions. + +1. Check `AGENTS.md` in your working directory for team context. +2. Check `RESEARCH/`, `GUIDES/`, and `PLANS/` before searching externally. ## Workspace Layout diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 35aaec188d..73fc99e422 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -361,7 +361,8 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_NO_IGNORE_SELF")] pub no_ignore_self: bool, - /// Maximum number of context messages to include for thread replies and DMs. + /// Maximum number of context messages to include for channels, thread replies, + /// and DMs. /// Set to 0 to disable automatic context fetching. Max 100. #[arg(long, env = "BUZZ_ACP_CONTEXT_MESSAGE_LIMIT", default_value_t = 12, value_parser = clap::value_parser!(u32).range(0..=100))] diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index fa348eeb3c..1d9ce1900a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2326,6 +2326,20 @@ async fn tokio_main() -> Result<()> { continue; } + // Relay-synthesized state overlays support clients and + // context fetches; they are never user or agent work. + // Dropping them here prevents wildcard subscriptions + // from turning a thread-summary update into a model + // turn whose reply produces another summary update. + if is_non_promptable_relay_state_kind(kind_u32) { + tracing::debug!( + channel_id = %buzz_event.channel_id, + kind = kind_u32, + "dropping relay-only state event before prompt" + ); + continue; + } + if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex { tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event"); continue; @@ -3088,6 +3102,10 @@ fn is_owner_control_command( && event_mentions_agent(event, agent_pubkey_hex) } +fn is_non_promptable_relay_state_kind(kind_u32: u32) -> bool { + buzz_core::kind::is_relay_only_kind(kind_u32) +} + // ── signal_in_flight_task ───────────────────────────────────────────────────── /// Decide which [`ControlSignal`] (if any) to send to an in-flight turn when a @@ -4546,21 +4564,23 @@ fn build_mcp_servers(config: &Config) -> Vec { if config.mcp_command.is_empty() { return vec![]; } + let server_name = std::path::Path::new(&config.mcp_command) + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or("mcp") + .to_string(); + let can_sign_buzz_messages = server_name == "buzz-message-mcp"; vec![McpServer { - name: std::path::Path::new(&config.mcp_command) - .file_stem() - .and_then(|s| s.to_str()) - .unwrap_or("mcp") - .to_string(), + name: server_name, command: config.mcp_command.clone(), args: vec![], env: { - let mut env = vec![ - EnvVar { - name: "BUZZ_RELAY_URL".into(), - value: config.relay_url.clone(), - }, - EnvVar { + let mut env = vec![EnvVar { + name: "BUZZ_RELAY_URL".into(), + value: config.relay_url.clone(), + }]; + if can_sign_buzz_messages { + env.push(EnvVar { name: "BUZZ_PRIVATE_KEY".into(), // bech32 encoding of a valid secret key is infallible. // Panic here is correct: injecting a bogus secret would cause @@ -4570,16 +4590,16 @@ fn build_mcp_servers(config: &Config) -> Vec { .secret_key() .to_bech32() .expect("secret key bech32 encoding should never fail"), - }, - ]; - // Forward BUZZ_AUTH_TAG (NIP-OA owner attestation credential) - // so the MCP server can attach it to every signed event. - if let Ok(auth_tag) = std::env::var("BUZZ_AUTH_TAG") { - if !auth_tag.is_empty() { - env.push(EnvVar { - name: "BUZZ_AUTH_TAG".into(), - value: auth_tag, - }); + }); + // Forward BUZZ_AUTH_TAG (NIP-OA owner attestation credential) + // only to the restricted messaging personality. + if let Ok(auth_tag) = std::env::var("BUZZ_AUTH_TAG") { + if !auth_tag.is_empty() { + env.push(EnvVar { + name: "BUZZ_AUTH_TAG".into(), + value: auth_tag, + }); + } } } // Forward the agent's display name so dev-mcp can use it as the git @@ -5161,6 +5181,45 @@ mod author_gate_tests { } } +#[cfg(test)] +mod relay_state_gate_tests { + use super::*; + use buzz_core::kind::{ + KIND_CHANNEL_SUMMARY, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, + KIND_PRESENCE_SNAPSHOT, KIND_THREAD_SUMMARY, KIND_WINDOW_BOUNDS, + }; + + #[test] + fn relay_state_overlays_never_open_model_turns() { + for kind in [ + KIND_CHANNEL_SUMMARY, + KIND_PRESENCE_SNAPSHOT, + KIND_THREAD_SUMMARY, + KIND_WINDOW_BOUNDS, + ] { + assert!( + is_non_promptable_relay_state_kind(kind), + "relay state kind {kind} must be dropped before prompting" + ); + } + } + + #[test] + fn work_and_membership_notifications_retain_existing_paths() { + for kind in [ + KIND_STREAM_MESSAGE, + KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, + ] { + assert!( + !is_non_promptable_relay_state_kind(kind), + "work or membership kind {kind} must not be swallowed by relay state gating" + ); + } + } +} + #[cfg(test)] mod observer_snapshot_race_tests { use super::*; @@ -6215,7 +6274,7 @@ mod build_mcp_servers_tests { relay_url: "ws://localhost:3000".into(), agent_command: "goose".into(), agent_args: vec!["acp".into()], - mcp_command: "test-mcp-server".into(), + mcp_command: "buzz-message-mcp".into(), idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS, max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS, agents: 1, @@ -6261,7 +6320,7 @@ mod build_mcp_servers_tests { let servers = build_mcp_servers(&config); assert_eq!(servers.len(), 1); let server = &servers[0]; - assert_eq!(server.name, "test-mcp-server"); + assert_eq!(server.name, "buzz-message-mcp"); let names: Vec<&str> = server.env.iter().map(|e| e.name.as_str()).collect(); assert!( @@ -6304,6 +6363,25 @@ mod build_mcp_servers_tests { assert!(!has_auth_tag, "empty BUZZ_AUTH_TAG should not be forwarded"); } + #[test] + fn general_mcp_server_never_receives_signing_credentials() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("BUZZ_AUTH_TAG", "test-attestation-tag"); + let mut config = test_config(); + config.mcp_command = "buzz-dev-mcp".into(); + let servers = build_mcp_servers(&config); + std::env::remove_var("BUZZ_AUTH_TAG"); + + let names: Vec<&str> = servers[0] + .env + .iter() + .map(|entry| entry.name.as_str()) + .collect(); + assert!(!names.contains(&"BUZZ_PRIVATE_KEY")); + assert!(!names.contains(&"BUZZ_AUTH_TAG")); + assert!(names.contains(&"BUZZ_RELAY_URL")); + } + #[test] fn test_display_name_set_is_forwarded_to_mcp_server() { let _guard = ENV_LOCK.lock().unwrap(); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 33bd5507fb..9262d18f63 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -576,11 +576,11 @@ pub struct PromptContext { /// (`include_str!`) is inherently `'static`. pub base_prompt: Option<&'static str>, pub cwd: String, - /// REST client for pre-prompt context fetches (thread/DM history). + /// REST client for pre-prompt context fetches (channel/thread/DM history). pub rest_client: RestClient, /// Shared channel metadata for startup-known and dynamically joined channels. pub channel_info: ChannelInfoResolver, - /// Max messages to include in thread/DM context. 0 = disabled. + /// Max messages to include in channel/thread/DM context. 0 = disabled. pub context_message_limit: u32, /// Max turns per session before proactive rotation. 0 = disabled. pub max_turns_per_session: u32, @@ -1438,6 +1438,41 @@ fn send_prompt_result( }); } +async fn publish_harness_fallback( + agent: &mut OwnedAgent, + batch: Option<&FlushBatch>, + source: &PromptSource, + ctx: &PromptContext, +) { + let (fallback_reply, secure_message_called) = agent.acp.take_turn_delivery(); + if secure_message_called + || fallback_reply.trim().is_empty() + || !matches!(source, PromptSource::Channel(_)) + { + return; + } + let Some(batch) = batch else { + return; + }; + let thread_tags = batch + .events + .last() + .map(|event| crate::queue::parse_thread_tags(&event.event)) + .unwrap_or_default(); + tracing::warn!( + target: "pool::delivery", + channel = %batch.channel_id, + "agent returned text without secure send_message call; publishing harness fallback" + ); + post_failure_notice( + &ctx.rest_client, + batch.channel_id, + &thread_tags, + fallback_reply.trim(), + ) + .await; +} + /// Core async function spawned for each prompt. /// /// Lifecycle: @@ -2045,6 +2080,7 @@ pub async fn run_prompt_task( team_instructions: standing.team_instructions, agent_canvas: standing.agent_canvas, standing_context_sent, + secure_messaging: has_secure_buzz_messaging(&ctx.mcp_servers), }, ) } else { @@ -2269,6 +2305,13 @@ pub async fn run_prompt_task( &source, &control_signal, ); + publish_harness_fallback( + &mut agent, + batch.as_ref(), + &source, + &ctx, + ) + .await; let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( &ctx, @@ -2309,6 +2352,8 @@ pub async fn run_prompt_task( agent.state.heartbeat_standing_context_sent = true; } + publish_harness_fallback(&mut agent, batch.as_ref(), &source, &ctx).await; + let should_rotate = matches!( stop_reason, StopReason::MaxTokens | StopReason::MaxTurnRequests @@ -2806,7 +2851,8 @@ pub(crate) fn render_canvas_section(event_id: &str, timestamp: &str, channel_uui fn conversation_context_event_ids(context: Option<&ConversationContext>) -> HashSet { match context { Some(ConversationContext::Thread { messages, .. }) - | Some(ConversationContext::Dm { messages, .. }) => messages + | Some(ConversationContext::Dm { messages, .. }) + | Some(ConversationContext::Channel { messages, .. }) => messages .iter() .filter(|message| !message.event_id.is_empty()) .map(|message| message.event_id.clone()) @@ -2861,13 +2907,22 @@ fn conversation_context_delta( truncated, }) } + ConversationContext::Channel { + messages, + truncated, + } => { + let messages = filter(messages); + (!messages.is_empty()).then_some(ConversationContext::Channel { + messages, + truncated, + }) + } } } -/// Fetch conversation context (thread or DM) for a batch before prompting. +/// Fetch conversation context (thread, DM, or channel) for a batch before prompting. /// /// Returns `None` if: -/// - The event is a plain channel message (not a thread reply, not a DM) /// - The REST fetch fails or times out (graceful degradation) /// - `context_message_limit` is 0 /// @@ -2905,7 +2960,15 @@ async fn fetch_conversation_context( return fetch_dm_context(batch.channel_id, limit, &ctx.rest_client).await; } - None + // Plain channel message: fetch recent top-level channel history. + fetch_channel_context(batch.channel_id, limit, &ctx.rest_client).await +} + +/// Whether the configured MCP servers include the send-only buzz-message-mcp. +pub(crate) fn has_secure_buzz_messaging(mcp_servers: &[McpServer]) -> bool { + mcp_servers + .iter() + .any(|server| server.name == "buzz-message-mcp") } /// Normalize AND validate a pubkey for the batch profile API request. @@ -2938,7 +3001,8 @@ fn collect_prompt_pubkeys( let context_messages = match conversation_context { Some(ConversationContext::Thread { messages, .. }) - | Some(ConversationContext::Dm { messages, .. }) => Some(messages), + | Some(ConversationContext::Dm { messages, .. }) + | Some(ConversationContext::Channel { messages, .. }) => Some(messages), None => None, }; @@ -3280,6 +3344,158 @@ async fn fetch_dm_context( .await } +/// Fetch recent channel messages via Nostr query: stream messages by `#h` tag. +async fn fetch_channel_context( + channel_id: Uuid, + limit: u32, + rest: &RestClient, +) -> Option { + let filter = channel_context_filter(channel_id, limit); + + fetch_with_retry(|| async { + match timeout( + CONTEXT_FETCH_TIMEOUT, + rest.query_raw_filters(std::slice::from_ref(&filter)), + ) + .await + { + Ok(Ok(json)) => parse_nostr_channel_response(json, channel_id, limit), + Ok(Err(e)) => { + tracing::warn!( + channel_id = %channel_id, + "channel context fetch failed: {e} — will retry" + ); + None + } + Err(_) => { + tracing::warn!( + channel_id = %channel_id, + "channel context fetch timed out — will retry" + ); + None + } + } + }) + .await +} + +/// Build the authenticated bridge's server-side top-level channel-window query. +/// +/// Vanilla Nostr filters cannot express "no reply e-tag". The relay's frozen +/// `top_level` extension applies that predicate before the row limit and emits +/// a kind-39006 bounds overlay with the authoritative `has_more` result. +fn channel_context_filter(channel_id: Uuid, limit: u32) -> serde_json::Value { + serde_json::json!({ + "kinds": [ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_STREAM_MESSAGE_V2, + ], + "#h": [channel_id.to_string()], + "limit": limit, + "top_level": true, + }) +} + +/// Whether a Nostr event JSON value is a supported stream-message kind. +fn is_supported_stream_message_kind(kind: u32) -> bool { + kind == buzz_core::kind::KIND_STREAM_MESSAGE || kind == buzz_core::kind::KIND_STREAM_MESSAGE_V2 +} + +/// Extract the `#h` channel tag from a Nostr event JSON value. +fn event_channel_tag(ev: &serde_json::Value) -> Option<&str> { + ev.get("tags")?.as_array()?.iter().find_map(|tag| { + let parts = tag.as_array()?; + if parts.first()?.as_str()? == "h" && parts.len() >= 2 { + parts[1].as_str() + } else { + None + } + }) +} + +/// Whether an event carries a Nostr thread/reference tag. +fn event_has_thread_tag(ev: &serde_json::Value) -> bool { + ev.get("tags") + .and_then(|tags| tags.as_array()) + .is_some_and(|tags| { + tags.iter().any(|tag| { + tag.as_array() + .and_then(|parts| parts.first()) + .and_then(|part| part.as_str()) + == Some("e") + }) + }) +} + +/// Read the relay-authored channel-window exhaustion fact. +fn channel_window_has_more(events: &[serde_json::Value], channel_id: Uuid) -> Option { + let channel_str = channel_id.to_string(); + events.iter().find_map(|event| { + let kind = event.get("kind")?.as_u64()? as u32; + if kind != buzz_core::kind::KIND_WINDOW_BOUNDS + || event_channel_tag(event) != Some(channel_str.as_str()) + { + return None; + } + let content = event.get("content")?.as_str()?; + serde_json::from_str::(content) + .ok()? + .get("has_more")? + .as_bool() + }) +} + +/// Parse a Nostr query response (array of events) into channel context. +/// +/// Validates channel, kind, and top-level status; sorts chronologically (oldest +/// first); and caps at `limit` messages (keeping the most recent). +fn parse_nostr_channel_response( + json: serde_json::Value, + channel_id: Uuid, + limit: u32, +) -> Option { + let events = json.as_array()?; + // Missing/malformed bounds cannot prove exhaustion. Mark the display + // conservatively so it never claims the supplied window is complete. + let relay_has_more = channel_window_has_more(events, channel_id).unwrap_or(true); + let channel_str = channel_id.to_string(); + + let mut messages: Vec<(u64, ContextMessage)> = events + .iter() + .filter(|ev| { + let kind = ev.get("kind").and_then(|v| v.as_u64()).unwrap_or(0) as u32; + if !is_supported_stream_message_kind(kind) { + return false; + } + event_channel_tag(ev) == Some(channel_str.as_str()) && !event_has_thread_tag(ev) + }) + .filter_map(|ev| { + let ts = ev.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0); + json_to_context_message(ev).map(|msg| (ts, msg)) + }) + .collect(); + + messages.sort_by_key(|(ts, _)| *ts); + + let valid_count = messages.len(); + if messages.len() > limit as usize { + let skip = messages.len() - limit as usize; + messages = messages.split_off(skip); + } + + let messages: Vec = messages.into_iter().map(|(_, msg)| msg).collect(); + let truncated = valid_count > limit as usize || relay_has_more; + + if messages.is_empty() { + return None; + } + + Some(ConversationContext::Channel { + messages, + truncated, + }) +} + /// Parse the legacy REST thread response (used in tests only). #[cfg(test)] fn parse_thread_response(json: serde_json::Value) -> Option { @@ -4776,6 +4992,247 @@ mod tests { assert!(parse_dm_response(json, 12).is_none()); } + #[test] + fn test_parse_nostr_channel_response_chronological_order() { + let channel_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(); + let json = json!([ + { + "id": "0000000000000000000000000000000000000000000000000000000000000002", + "pubkey": "pub2", + "kind": 9, + "content": "newer", + "created_at": 200, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000001", + "pubkey": "pub1", + "kind": 9, + "content": "older", + "created_at": 100, + "tags": [["h", channel_id.to_string()]] + } + ]); + + let ctx = parse_nostr_channel_response(json, channel_id, 12).expect("should parse"); + match ctx { + ConversationContext::Channel { messages, .. } => { + assert_eq!(messages.len(), 2); + assert_eq!(messages[0].content, "older"); + assert_eq!(messages[1].content, "newer"); + } + _ => panic!("expected Channel context"), + } + } + + #[test] + fn test_parse_nostr_channel_response_filters_wrong_channel_kind_and_thread() { + let channel_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(); + let other_channel = Uuid::parse_str("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb").unwrap(); + let json = json!([ + { + "id": "0000000000000000000000000000000000000000000000000000000000000001", + "pubkey": "pub1", + "kind": 9, + "content": "valid", + "created_at": 100, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000002", + "pubkey": "pub2", + "kind": 9, + "content": "wrong channel", + "created_at": 200, + "tags": [["h", other_channel.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000003", + "pubkey": "pub3", + "kind": 39000, + "content": "metadata only", + "created_at": 300, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000004", + "pubkey": "pub4", + "kind": 9, + "content": "thread reply", + "created_at": 400, + "tags": [ + ["h", channel_id.to_string()], + ["e", "root-event-id", "", "reply"] + ] + } + ]); + + let ctx = parse_nostr_channel_response(json, channel_id, 12).expect("should parse"); + match ctx { + ConversationContext::Channel { messages, .. } => { + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].content, "valid"); + } + _ => panic!("expected Channel context"), + } + } + + #[test] + fn test_parse_nostr_channel_response_truncates_to_limit() { + let channel_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(); + let json = json!([ + { + "id": "0000000000000000000000000000000000000000000000000000000000000001", + "pubkey": "pub1", + "kind": 9, + "content": "oldest", + "created_at": 100, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000002", + "pubkey": "pub2", + "kind": 9, + "content": "middle", + "created_at": 200, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000003", + "pubkey": "pub3", + "kind": 9, + "content": "newest", + "created_at": 300, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000004", + "pubkey": "relay", + "kind": 39006, + "content": serde_json::json!({ + "has_more": true, + "next_cursor": {"created_at": 100, "id": "cursor"} + }).to_string(), + "created_at": 301, + "tags": [["h", channel_id.to_string()]] + } + ]); + + let ctx = parse_nostr_channel_response(json, channel_id, 2).expect("should parse"); + match ctx { + ConversationContext::Channel { + messages, + truncated, + } => { + assert_eq!(messages.len(), 2); + assert_eq!(messages[0].content, "middle"); + assert_eq!(messages[1].content, "newest"); + assert!(truncated); + } + _ => panic!("expected Channel context"), + } + } + + #[test] + fn test_parse_nostr_channel_response_exact_limit_is_not_fabricated_truncation() { + let channel_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(); + let json = json!([ + { + "id": "0000000000000000000000000000000000000000000000000000000000000001", + "pubkey": "pub1", + "kind": 9, + "content": "older", + "created_at": 100, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000002", + "pubkey": "pub2", + "kind": 9, + "content": "newer", + "created_at": 200, + "tags": [["h", channel_id.to_string()]] + }, + { + "id": "0000000000000000000000000000000000000000000000000000000000000003", + "pubkey": "relay", + "kind": 39006, + "content": serde_json::json!({ + "has_more": false, + "next_cursor": null + }).to_string(), + "created_at": 201, + "tags": [["h", channel_id.to_string()]] + } + ]); + + let ctx = parse_nostr_channel_response(json, channel_id, 2).expect("should parse"); + match ctx { + ConversationContext::Channel { + messages, + truncated, + } => { + assert_eq!(messages.len(), 2); + assert!(!truncated); + } + _ => panic!("expected Channel context"), + } + } + + #[test] + fn test_channel_context_filter_uses_server_side_top_level_window() { + let channel_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(); + let filter = channel_context_filter(channel_id, 12); + + assert_eq!(filter["#h"], json!([channel_id.to_string()])); + assert_eq!( + filter["kinds"], + json!([ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_STREAM_MESSAGE_V2 + ]) + ); + assert_eq!(filter["limit"], json!(12)); + assert_eq!(filter["top_level"], json!(true)); + } + + #[test] + fn test_parse_nostr_channel_response_missing_bounds_is_conservative() { + let channel_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(); + let json = json!([{ + "id": "0000000000000000000000000000000000000000000000000000000000000001", + "pubkey": "pub1", + "kind": 9, + "content": "message", + "created_at": 100, + "tags": [["h", channel_id.to_string()]] + }]); + + let ctx = parse_nostr_channel_response(json, channel_id, 12).expect("should parse"); + match ctx { + ConversationContext::Channel { truncated, .. } => assert!(truncated), + _ => panic!("expected Channel context"), + } + } + + #[test] + fn test_has_secure_buzz_messaging_detects_message_mcp() { + use crate::acp::McpServer; + assert!(!has_secure_buzz_messaging(&[])); + assert!(has_secure_buzz_messaging(&[McpServer { + name: "buzz-message-mcp".into(), + command: "buzz-message-mcp".into(), + args: vec![], + env: vec![], + }])); + assert!(!has_secure_buzz_messaging(&[McpServer { + name: "buzz-dev-mcp".into(), + command: "buzz-dev-mcp".into(), + args: vec![], + env: vec![], + }])); + } + #[test] fn test_parse_dm_response_missing_messages_key() { let json = json!({ "data": [] }); diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 3bf1962242..d2eda5a909 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -985,6 +985,11 @@ pub enum ConversationContext { total: usize, truncated: bool, }, + /// Recent top-level channel messages (not a thread reply, not a DM). + Channel { + messages: Vec, + truncated: bool, + }, } /// A single message in a conversation context section. @@ -1151,17 +1156,27 @@ pub(crate) fn format_event_block( /// Append a reply instruction when the agent is responding to a thread event. /// -/// Tells the agent to default to `--reply-to ` for ordinary replies -/// while still allowing an explicit human request to post at the channel root or -/// top level. -fn append_reply_instruction(s: &mut String, event_id: &str) { - s.push_str(&format!( - "\nIMPORTANT: For ordinary replies in this turn, use `--reply-to {event_id}` \ - on `buzz messages send` so the conversation stays threaded. \ - If the human explicitly asks for a channel-root, top-level, \ - or broadcast post, send that message without `--reply-to`. \ - If the requested destination is ambiguous, ask before sending." - )); +/// Secure agents receive a tool argument, never a terminal command. Other +/// agents retain the CLI instruction. +fn append_reply_instruction(s: &mut String, event_id: &str, secure_messaging: bool) { + if secure_messaging { + s.push_str(&format!( + "\nIMPORTANT: For ordinary replies in this turn, call \ + `mcp__buzz_message_mcp__send_message` with its `reply_to` argument \ + set to `{event_id}` so the conversation stays threaded. If the \ + human explicitly asks for a channel-root, top-level, or broadcast \ + post, omit `reply_to`. If the requested destination is ambiguous, \ + ask before sending." + )); + } else { + s.push_str(&format!( + "\nIMPORTANT: For ordinary replies in this turn, use `--reply-to {event_id}` \ + on `buzz messages send` so the conversation stays threaded. \ + If the human explicitly asks for a channel-root, top-level, \ + or broadcast post, send that message without `--reply-to`. \ + If the requested destination is ambiguous, ask before sending." + )); + } } /// Append a new-thread reply instruction for a human-facing top-level mention. @@ -1169,14 +1184,25 @@ fn append_reply_instruction(s: &mut String, event_id: &str) { /// The triggering mention has no thread tags, so the agent's reply becomes the /// thread root. Anchoring to the triggering event (rather than leaving the /// choice open) prevents replying into a stale/unrelated prior thread. -fn append_new_thread_reply_instruction(s: &mut String, event_id: &str) { - s.push_str(&format!( - "\nIMPORTANT: This is a new top-level message. For ordinary replies in \ - this turn, use `--reply-to {event_id}` on `buzz messages send` — the \ - triggering message is the thread root. Do NOT reply into any other \ - (older) thread. If the human explicitly asks for a channel-root, \ - top-level, or broadcast post, send that message without `--reply-to`." - )); +fn append_new_thread_reply_instruction(s: &mut String, event_id: &str, secure_messaging: bool) { + if secure_messaging { + s.push_str(&format!( + "\nIMPORTANT: This is a new top-level message. For ordinary replies \ + in this turn, call `mcp__buzz_message_mcp__send_message` with its \ + `reply_to` argument set to `{event_id}` — the triggering message \ + is the thread root. Do NOT reply into any other (older) thread. \ + If the human explicitly asks for a channel-root, top-level, or \ + broadcast post, omit `reply_to`." + )); + } else { + s.push_str(&format!( + "\nIMPORTANT: This is a new top-level message. For ordinary replies in \ + this turn, use `--reply-to {event_id}` on `buzz messages send` — the \ + triggering message is the thread root. Do NOT reply into any other \ + (older) thread. If the human explicitly asks for a channel-root, \ + top-level, or broadcast post, send that message without `--reply-to`." + )); + } } /// Decide whether a turn is human-facing for reply-anchor purposes. @@ -1231,6 +1257,14 @@ fn resolve_reply_anchor( ) } +/// Hint text when secure messaging (`buzz-message-mcp`) is active. +const SECURE_CONTEXT_HINT: &str = + "Conversation context is harness-supplied; use the send_message tool for replies."; + +/// Hint text when secure messaging is active and context was fetched. +const SECURE_CONTEXT_INCLUDED_HINT: &str = + "Channel context included below. Conversation history is harness-supplied; use the send_message tool for replies."; + /// Format a `[Context]` hints section based on event scope. /// /// `reply_anchor` is the pre-resolved `--reply-to` target for this turn (see @@ -1246,6 +1280,7 @@ fn format_context_hints( has_conversation_context: bool, conversation_context_had_delivered_events: bool, reply_anchor: Option<&str>, + secure_messaging: bool, ) -> String { let channel_display = match channel_info { Some(ci) => format!("{} (#{channel_id})", ci.name), @@ -1258,7 +1293,15 @@ fn format_context_hints( let is_reply = thread_tags.root_event_id.is_some(); // DM replies use thread command because /messages excludes thread replies. // DM non-replies use get for recent conversation. - let ctx_hint = if has_conversation_context && is_reply { + let ctx_hint = if secure_messaging { + if has_conversation_context && is_reply { + "Thread context included below. Conversation history is harness-supplied; use the send_message tool for replies." + } else if has_conversation_context { + "Conversation context included below. Conversation history is harness-supplied; use the send_message tool for replies." + } else { + SECURE_CONTEXT_HINT + } + } else if has_conversation_context && is_reply { "Thread context included below. Use `buzz messages thread --channel --event ` for full history if truncated." } else if has_conversation_context { "Conversation context included below. Use `buzz messages get --channel ` for full history if truncated." @@ -1286,12 +1329,18 @@ fn format_context_hints( } } if let Some(event_id) = reply_anchor { - append_reply_instruction(&mut s, event_id); + append_reply_instruction(&mut s, event_id, secure_messaging); } } s } else if let Some(ref root) = thread_tags.root_event_id { - let ctx_hint = if has_conversation_context { + let ctx_hint = if secure_messaging { + if has_conversation_context { + "Thread context included below. Conversation history is harness-supplied; use the send_message tool for replies." + } else { + SECURE_CONTEXT_HINT + } + } else if has_conversation_context { "Thread context included below. Use `buzz messages thread --channel --event ` for full history if truncated." } else if conversation_context_had_delivered_events { "Earlier thread context was already delivered in this session. Use `buzz messages thread --channel --event ` to re-read it." @@ -1311,18 +1360,29 @@ fn format_context_hints( } s.push_str(&format!("\n{ctx_hint}")); if let Some(event_id) = reply_anchor { - append_reply_instruction(&mut s, event_id); + append_reply_instruction(&mut s, event_id, secure_messaging); } s } else { + let ctx_hint = if secure_messaging { + if has_conversation_context { + SECURE_CONTEXT_INCLUDED_HINT + } else { + SECURE_CONTEXT_HINT + } + } else if has_conversation_context { + "Channel context included below. Use `buzz messages get --channel ` for full history if truncated." + } else { + "Use `buzz messages get --channel ` for recent messages if needed." + }; let mut s = format!( "[Context]\n\ Scope: channel\n\ Channel: {channel_display}\n\ - Hint: Use `buzz messages get --channel ` for recent messages if needed." + Hint: {ctx_hint}" ); if let Some(event_id) = reply_anchor { - append_new_thread_reply_instruction(&mut s, event_id); + append_new_thread_reply_instruction(&mut s, event_id, secure_messaging); } s } @@ -1333,24 +1393,52 @@ fn format_conversation_context( ctx: &ConversationContext, profile_lookup: Option<&PromptProfileLookup>, ) -> String { - let (label, messages, total, truncated) = match ctx { + let (header, messages) = match ctx { ConversationContext::Thread { messages, total, truncated, - } => ("Thread Context", messages, total, truncated), + } => { + let trunc_label = if *truncated { ", truncated" } else { "" }; + ( + format!( + "[Thread Context ({} of {total} messages{trunc_label})]", + messages.len() + ), + messages, + ) + } ConversationContext::Dm { messages, total, truncated, - } => ("Conversation Context", messages, total, truncated), + } => { + let trunc_label = if *truncated { ", truncated" } else { "" }; + ( + format!( + "[Conversation Context ({} of {total} messages{trunc_label})]", + messages.len() + ), + messages, + ) + } + ConversationContext::Channel { + messages, + truncated, + } => { + let qualifier = if *truncated { + format!( + "latest {} messages, older history may be omitted", + messages.len() + ) + } else { + format!("{} messages", messages.len()) + }; + (format!("[Channel Context ({qualifier})]"), messages) + } }; - let trunc_label = if *truncated { ", truncated" } else { "" }; - let mut s = format!( - "[{label} ({} of {total} messages{trunc_label})]", - messages.len() - ); + let mut s = header; for (i, msg) in messages.iter().enumerate() { s.push_str(&format!( "\n[{}] {} ({}): {}", @@ -1395,6 +1483,9 @@ pub struct FormatPromptArgs<'a> { /// Defaults to `false` so a caller that never sets it behaves as if this /// were the session's first message. pub standing_context_sent: bool, + /// When true, the agent has `buzz-message-mcp` (send-only) and must not + /// be directed to CLI history commands (`buzz messages get`, etc.). + pub secure_messaging: bool, } /// The prompt sections that do not change for the life of a session: base @@ -1541,6 +1632,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec> = Mutex::new(None); + +/// Serializes HTTP bridge requests so prompt-task clones cannot pass the gate +/// together and stampede the relay before the first 429 arms it. +static HTTP_BRIDGE_PERMITS: Semaphore = Semaphore::const_new(1); + +/// Parse an HTTP `Retry-After` header value as whole seconds. +/// +/// Integer seconds are returned directly. HTTP-date values and garbage input +/// return `None` so callers can apply the conservative default. +pub(crate) fn parse_http_retry_after_secs(header: Option<&str>) -> Option { + let value = header?.trim(); + if value.is_empty() { + return None; + } + value.parse::().ok() +} + +/// Arm or extend the shared HTTP rate-limit gate from a 429 `Retry-After` hint. +/// +/// Hints below 2 s (including absent/`0`) floor to [`HTTP_RATE_LIMIT_DEFAULT_SECS`]. +/// Overlapping 429s only extend the deadline — a shorter hint never shortens an +/// active gate. +fn activate_http_rate_limit(retry_secs: Option) { + let secs = match retry_secs { + Some(s) if s >= 2 => s.min(HTTP_RATE_LIMIT_MAX_HINT_SECS), + _ => HTTP_RATE_LIMIT_DEFAULT_SECS, + }; + let new_deadline = tokio::time::Instant::now() + jittered_duration(Duration::from_secs(secs)); + let mut guard = HTTP_RATE_GATE.lock().unwrap_or_else(|e| e.into_inner()); + match *guard { + Some(existing) if existing > new_deadline => {} + _ => *guard = Some(new_deadline), + } +} + +/// Wait until the shared HTTP rate-limit gate is clear. +/// +/// Returns immediately when inactive. Loops after sleeping because a concurrent +/// 429 may extend the expiry while this caller is parked. The mutex is not held +/// across the sleep. +async fn wait_for_http_rate_limit() { + loop { + let expiry = { + let mut guard = HTTP_RATE_GATE.lock().unwrap_or_else(|e| e.into_inner()); + match *guard { + Some(expiry) if expiry > tokio::time::Instant::now() => Some(expiry), + _ => { + *guard = None; + None + } + } + }; + match expiry { + Some(expiry) => sleep_until(expiry).await, + None => return, + } + } +} + +/// Reset the process-local HTTP rate gate. Test-only. +#[cfg(test)] +pub(crate) fn reset_http_rate_gate_for_test() { + *HTTP_RATE_GATE.lock().unwrap_or_else(|e| e.into_inner()) = None; +} + /// Lightweight HTTP client for pre-prompt context fetches via the Nostr HTTP bridge. /// /// Extracted from `HarnessRelay` fields so it can be shared (via `Arc`) with @@ -313,6 +391,8 @@ impl RestClient { /// Retry helper: executes `build_request` up to 4 times (1 attempt + 3 retries) /// on transient failures (429, 502, 503, 504, timeout, connect errors). /// + /// HTTP 429 responses arm a process-local gate from `Retry-After` (shared by + /// every [`RestClient`] clone) instead of using the short fixed backoff. /// NIP-98 auth events are re-signed on each attempt (they have a ±60s window). async fn request_with_retry( &self, @@ -325,22 +405,49 @@ impl RestClient { Fut: std::future::Future>, { let mut last_err = None; + let mut last_was_rate_limited = false; for (attempt, delay) in std::iter::once(None) .chain(REST_RETRY_BASE_DELAYS.iter().map(|d| Some(*d))) .enumerate() { if let Some(base) = delay { - let jittered = jittered_duration(base); - tracing::debug!( - "retrying {method} {path} (attempt {attempt}) in {:.1}s", - jittered.as_secs_f64() - ); - tokio::time::sleep(jittered).await; + if !last_was_rate_limited { + let jittered = jittered_duration(base); + tracing::debug!( + "retrying {method} {path} (attempt {attempt}) in {:.1}s", + jittered.as_secs_f64() + ); + tokio::time::sleep(jittered).await; + } } + last_was_rate_limited = false; + + let _permit = HTTP_BRIDGE_PERMITS + .acquire() + .await + .map_err(|_| RelayError::Http("HTTP bridge semaphore closed".into()))?; + // Check the gate only after admission. A waiter may have entered the + // semaphore queue before another request armed the gate. + wait_for_http_rate_limit().await; match build_request().await { Ok(resp) if resp.status().is_success() => return Ok(resp), + Ok(resp) if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS => { + let retry_hdr = resp + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|v| v.to_str().ok()); + let retry_secs = parse_http_retry_after_secs(retry_hdr); + activate_http_rate_limit(retry_secs); + tracing::warn!( + "{method} {path} returned HTTP 429 — shared gate armed (retry_after={retry_secs:?})" + ); + last_err = Some(RelayError::Http(format!( + "{method} {path} returned HTTP 429" + ))); + last_was_rate_limited = true; + } Ok(resp) if is_retriable_status(resp.status()) => { let status = resp.status(); tracing::warn!("{method} {path} returned retriable HTTP {status}"); @@ -421,6 +528,20 @@ impl RestClient { .map_err(|e| RelayError::Http(e.to_string())) } + /// Query events through a relay-specific raw filter extension. + /// + /// `nostr::Filter` intentionally drops unknown fields, so callers that use + /// authenticated bridge extensions such as `top_level` must preserve the + /// raw JSON object on the wire. + pub async fn query_raw_filters(&self, filters: &[Value]) -> Result { + let body_bytes = serde_json::to_vec(filters) + .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?; + let resp = self.bridge_post("/query", &body_bytes).await?; + resp.json() + .await + .map_err(|e| RelayError::Http(e.to_string())) + } + /// Submit a signed event via the HTTP bridge: `POST /events` with NIP-98 auth. /// /// The event must already be signed. Returns the relay response JSON. @@ -6246,4 +6367,132 @@ mod tests { "channel_dropped_since must be cleared on successful drain" ); } + + // ── HTTP bridge rate gate (shared across RestClient clones) ─────────────── + + /// Serialize gate tests — the gate is process-wide static state. + static HTTP_GATE_TEST_SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + + #[test] + fn parse_http_retry_after_secs_valid_and_invalid() { + assert_eq!(parse_http_retry_after_secs(Some("12")), Some(12)); + assert_eq!(parse_http_retry_after_secs(Some("0")), Some(0)); + assert_eq!(parse_http_retry_after_secs(None), None); + assert_eq!(parse_http_retry_after_secs(Some("")), None); + assert_eq!(parse_http_retry_after_secs(Some("not-a-number")), None); + assert_eq!( + parse_http_retry_after_secs(Some("Wed, 21 Oct 2015 07:28:00 GMT")), + None + ); + } + + #[tokio::test(start_paused = true)] + async fn http_rate_gate_extends_without_shortening() { + let _serial = HTTP_GATE_TEST_SERIAL.lock().await; + reset_http_rate_gate_for_test(); + + activate_http_rate_limit(Some(30)); + let first = HTTP_RATE_GATE.lock().unwrap().unwrap(); + + activate_http_rate_limit(Some(1)); + let second = HTTP_RATE_GATE.lock().unwrap().unwrap(); + assert!( + second >= first, + "shorter Retry-After must not shorten an active gate" + ); + + reset_http_rate_gate_for_test(); + } + + #[tokio::test(start_paused = true)] + async fn http_rate_gate_missing_retry_after_uses_default() { + let _serial = HTTP_GATE_TEST_SERIAL.lock().await; + reset_http_rate_gate_for_test(); + + let before = tokio::time::Instant::now(); + activate_http_rate_limit(None); + let deadline = HTTP_RATE_GATE.lock().unwrap().unwrap(); + let remaining = deadline.duration_since(before); + + // Default is 5s with up to +20% jitter → at least 4s. + assert!(remaining >= Duration::from_secs(4)); + assert!(remaining <= Duration::from_secs(6)); + + reset_http_rate_gate_for_test(); + } + + #[tokio::test] + async fn concurrent_rest_client_clones_share_http_rate_gate() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{Arc, Mutex as StdMutex}; + + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let _serial = HTTP_GATE_TEST_SERIAL.lock().await; + reset_http_rate_gate_for_test(); + + let server = MockServer::start().await; + let hits = Arc::new(AtomicUsize::new(0)); + let hits_for_mock = hits.clone(); + let hit_times = Arc::new(StdMutex::new(Vec::new())); + let hit_times_for_mock = hit_times.clone(); + + Mock::given(method("POST")) + .and(path("/query")) + .respond_with(move |_: &wiremock::Request| { + hit_times_for_mock + .lock() + .unwrap() + .push(tokio::time::Instant::now()); + let n = hits_for_mock.fetch_add(1, Ordering::SeqCst); + if n == 0 { + ResponseTemplate::new(429).insert_header("Retry-After", "2") + } else { + ResponseTemplate::new(200).set_body_json(serde_json::json!([])) + } + }) + .mount(&server) + .await; + + let base_client = RestClient { + http: reqwest::Client::new(), + base_url: server.uri(), + keys: nostr::Keys::generate(), + auth_tag_json: None, + }; + + let filters = [nostr::Filter::new().kind(nostr::Kind::Metadata)]; + let mut handles = Vec::new(); + for _ in 0..4 { + let client = base_client.clone(); + let filters = filters.clone(); + handles.push(tokio::spawn(async move { client.query(&filters).await })); + } + + for handle in handles { + let result = handle.await.expect("task join"); + assert!( + result.is_ok(), + "expected shared gate retry to succeed: {result:?}" + ); + } + + let total_hits = hits.load(Ordering::SeqCst); + assert!( + total_hits <= 6, + "without shared pacing clones would each retry independently (got {total_hits} hits)" + ); + assert!( + total_hits >= 2, + "expected at least one 429 and follow-up successes (got {total_hits} hits)" + ); + let times = hit_times.lock().unwrap(); + assert!( + times[1].duration_since(times[0]) >= Duration::from_millis(1500), + "the first queued clone bypassed the shared Retry-After gate" + ); + + reset_http_rate_gate_for_test(); + } } diff --git a/crates/buzz-dev-mcp/Cargo.toml b/crates/buzz-dev-mcp/Cargo.toml index 8b711b8063..3783cd370b 100644 --- a/crates/buzz-dev-mcp/Cargo.toml +++ b/crates/buzz-dev-mcp/Cargo.toml @@ -13,6 +13,10 @@ path = "src/lib.rs" name = "buzz-dev-mcp" path = "src/main.rs" +[[bin]] +name = "buzz-message-mcp" +path = "src/message_main.rs" + [dependencies] buzz-cli = { path = "../buzz-cli" } git-credential-nostr = { path = "../git-credential-nostr" } diff --git a/crates/buzz-dev-mcp/src/lib.rs b/crates/buzz-dev-mcp/src/lib.rs index 9b98974802..e4ec937a95 100644 --- a/crates/buzz-dev-mcp/src/lib.rs +++ b/crates/buzz-dev-mcp/src/lib.rs @@ -10,6 +10,7 @@ use rmcp::{ use std::path::Path; use std::sync::Arc; +mod message; mod paths; mod read_file; mod rg; @@ -27,6 +28,61 @@ struct DevMcp { tool_router: ToolRouter, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum McpProfile { + Developer, + Messaging, +} + +fn mcp_profile_for_command(command: &str) -> McpProfile { + if command == "buzz-message-mcp" { + McpProfile::Messaging + } else { + McpProfile::Developer + } +} + +#[derive(Clone)] +struct MessagingMcp { + state: Arc, + tool_router: ToolRouter, +} + +#[tool_router] +impl MessagingMcp { + fn new(state: Arc) -> Self { + Self { + state, + tool_router: Self::tool_router(), + } + } + + #[tool( + name = "send_message", + description = "Send one authenticated Buzz message. Credentials stay inside this restricted MCP server and are never exposed to terminal or execute-code tools." + )] + async fn send_message( + &self, + Parameters(p): Parameters, + ) -> Result { + message::send_message(&self.state, p).await + } +} + +#[tool_handler(router = self.tool_router)] +impl ServerHandler for MessagingMcp { + fn get_info(&self) -> ServerInfo { + ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + .with_server_info(rmcp::model::Implementation::new( + "buzz-message-mcp", + env!("CARGO_PKG_VERSION"), + )) + .with_instructions( + "Use send_message for Buzz replies. Signing credentials are isolated in this server.", + ) + } +} + #[tool_router] impl DevMcp { fn new(state: Arc) -> Self { @@ -176,15 +232,67 @@ async fn async_main(cmd: String) -> Result<(), Box> { .with_ansi(false) .init(); - let cwd = std::env::current_dir()?; let shim = shim::Shim::install()?; - let state = Arc::new(shell::SharedState::new(cwd, shim)?); - - let service = DevMcp::new(state).serve(stdio()).await?; - service.waiting().await?; + match mcp_profile_for_command(&cmd) { + McpProfile::Messaging => { + let state = Arc::new(message::MessagingState::new(shim)); + let service = MessagingMcp::new(state).serve(stdio()).await?; + service.waiting().await?; + } + McpProfile::Developer => { + let cwd = std::env::current_dir()?; + let state = Arc::new(shell::SharedState::new(cwd, shim)?); + let service = DevMcp::new(state).serve(stdio()).await?; + service.waiting().await?; + } + } Ok(()) } +#[cfg(test)] +mod messaging_profile_tests { + use super::*; + + #[test] + fn buzz_message_mcp_selects_the_restricted_messaging_profile() { + assert_eq!( + mcp_profile_for_command("buzz-message-mcp"), + McpProfile::Messaging + ); + assert_eq!( + mcp_profile_for_command("buzz-dev-mcp"), + McpProfile::Developer + ); + } + + #[test] + fn messaging_send_builds_only_the_fixed_buzz_cli_command() { + let args = message::send_message_args(&message::SendMessageParams { + channel: "channel-id".to_string(), + content: "HEALTH_ACK token".to_string(), + reply_to: Some("event-id".to_string()), + mentions: vec!["pubkey-a".to_string(), "pubkey-b".to_string()], + }); + assert_eq!( + args, + vec![ + "messages", + "send", + "--channel", + "channel-id", + "--content", + "-", + "--reply-to", + "event-id", + "--mention", + "pubkey-a", + "--mention", + "pubkey-b", + ] + ); + } +} + /// Suppress the console window that Windows otherwise allocates for every /// console-subsystem child process spawned from a non-console parent. /// No-op on non-Windows platforms. diff --git a/crates/buzz-dev-mcp/src/message.rs b/crates/buzz-dev-mcp/src/message.rs new file mode 100644 index 0000000000..3e54a8470e --- /dev/null +++ b/crates/buzz-dev-mcp/src/message.rs @@ -0,0 +1,123 @@ +use crate::shim::Shim; +use rmcp::model::{CallToolResult, Content}; +use rmcp::ErrorData; +use schemars::JsonSchema; +use serde::Deserialize; +use std::process::Stdio; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::AsyncWriteExt; +use tokio::process::Command; + +const SEND_TIMEOUT: Duration = Duration::from_secs(20); + +pub struct MessagingState { + _shim: Shim, + buzz_path: std::path::PathBuf, +} + +impl MessagingState { + pub fn new(shim: Shim) -> Self { + let buzz_path = shim.buzz_path.clone(); + Self { + _shim: shim, + buzz_path, + } + } +} + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct SendMessageParams { + /// Channel UUID from the current Buzz context. + pub channel: String, + /// Message body. + pub content: String, + /// Optional event ID to reply to. + #[serde(default)] + pub reply_to: Option, + /// Optional public keys to mention. + #[serde(default)] + pub mentions: Vec, +} + +pub fn send_message_args(p: &SendMessageParams) -> Vec { + let mut args = vec![ + "messages".to_string(), + "send".to_string(), + "--channel".to_string(), + p.channel.clone(), + "--content".to_string(), + "-".to_string(), + ]; + if let Some(reply_to) = p.reply_to.as_deref().filter(|value| !value.is_empty()) { + args.push("--reply-to".to_string()); + args.push(reply_to.to_string()); + } + for mention in p.mentions.iter().filter(|value| !value.is_empty()) { + args.push("--mention".to_string()); + args.push(mention.clone()); + } + args +} + +pub async fn send_message( + state: &Arc, + p: SendMessageParams, +) -> Result { + if p.channel.trim().is_empty() { + return Ok(CallToolResult::error(vec![Content::text( + "channel must not be empty", + )])); + } + if p.content.trim().is_empty() { + return Ok(CallToolResult::error(vec![Content::text( + "content must not be empty", + )])); + } + + let mut command = Command::new(&state.buzz_path); + command + .args(send_message_args(&p)) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let output = match tokio::time::timeout(SEND_TIMEOUT, async { + let mut child = command + .spawn() + .map_err(|error| ErrorData::internal_error(error.to_string(), None))?; + let mut stdin = child + .stdin + .take() + .ok_or_else(|| ErrorData::internal_error("failed to open buzz stdin", None))?; + stdin + .write_all(p.content.as_bytes()) + .await + .map_err(|error| ErrorData::internal_error(error.to_string(), None))?; + drop(stdin); + child + .wait_with_output() + .await + .map_err(|error| ErrorData::internal_error(error.to_string(), None)) + }) + .await + { + Ok(result) => result?, + Err(_) => { + return Ok(CallToolResult::error(vec![Content::text( + "buzz messages send timed out after 20 seconds", + )])); + } + }; + let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string(); + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + if output.status.success() { + Ok(CallToolResult::success(vec![Content::text(stdout)])) + } else { + let detail = if stderr.is_empty() { stdout } else { stderr }; + Ok(CallToolResult::error(vec![Content::text(format!( + "buzz messages send failed ({}): {detail}", + output.status + ))])) + } +} diff --git a/crates/buzz-dev-mcp/src/message_main.rs b/crates/buzz-dev-mcp/src/message_main.rs new file mode 100644 index 0000000000..e2883ab692 --- /dev/null +++ b/crates/buzz-dev-mcp/src/message_main.rs @@ -0,0 +1,3 @@ +fn main() -> Result<(), Box> { + buzz_dev_mcp::run() +} diff --git a/crates/buzz-dev-mcp/src/shim.rs b/crates/buzz-dev-mcp/src/shim.rs index cccf0e6eca..d7c98db186 100644 --- a/crates/buzz-dev-mcp/src/shim.rs +++ b/crates/buzz-dev-mcp/src/shim.rs @@ -19,6 +19,7 @@ pub struct Shim { _dir: TempDir, pub path_env: String, pub git_env: Vec<(String, String)>, + pub buzz_path: PathBuf, } impl Shim { @@ -38,6 +39,7 @@ impl Shim { ] { symlink(&self_exe, &dir.path().join(name))?; } + let buzz_path = dir.path().join("buzz"); let original = std::env::var_os("PATH").unwrap_or_default(); let mut entries = vec![PathBuf::from(dir.path())]; @@ -71,6 +73,7 @@ impl Shim { _dir: dir, path_env, git_env, + buzz_path, }) } } diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 7aa954ce8e..720dc44d75 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -534,6 +534,34 @@ pub fn run() { state .managed_agent_restore_pending .store(true, Ordering::Release); + + // Keep start-on-launch agents alive after the initial restore. + // A killed child can remain as a zombie until its parent calls + // try_wait(); the same restore path both reaps that child and + // starts the now-missing runtime. The pending gate prevents this + // loop from racing workspace relay/identity installation. + let keepalive_handle = app.handle().clone(); + tauri::async_runtime::spawn(async move { + use std::time::Duration; + use tauri::Manager; + loop { + tokio::time::sleep(Duration::from_secs(30)).await; + let state = keepalive_handle.state::(); + if state.managed_agent_restore_pending.load(Ordering::Acquire) { + continue; + } + if let Err(error) = managed_agents::restore_managed_agents_on_launch( + &keepalive_handle, + &state.shutdown_started, + ) + .await + { + eprintln!( + "buzz-desktop: managed-agent keepalive restore failed: {error}" + ); + } + } + }); } // Periodic sweep: reap orphaned agents from dead instances every 60s. diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index ec804869c4..0f7b6954db 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -33,6 +33,19 @@ pub use stop::{stop_managed_agent_process, stop_managed_agent_workspace_pair}; mod sweep; pub(crate) use sweep::sweep_untracked_bundle_harnesses; +fn effective_mcp_command<'a>( + record: &'a ManagedAgentRecord, + effective_agent_command: &str, +) -> &'a str { + let configured = record.mcp_command.trim(); + if !configured.is_empty() { + return configured; + } + known_acp_runtime(effective_agent_command) + .and_then(|runtime| runtime.mcp_command) + .unwrap_or("") +} + mod process; #[cfg(test)] use process::{ @@ -291,10 +304,7 @@ pub fn build_managed_agent_summary( env: Default::default(), } }); - let effective_mcp_command = known_acp_runtime(&descriptor.command) - .and_then(|r| r.mcp_command) - .unwrap_or("") - .to_string(); + let effective_mcp_command = effective_mcp_command(record, &descriptor.command).to_string(); Ok(ManagedAgentSummary { pubkey: record.pubkey.clone(), @@ -414,9 +424,9 @@ pub fn spawn_agent_child( let runtime_key = ManagedAgentRuntimeKey::new(record.pubkey.clone(), relay_url)?; // Resolve the effective harness (agent command) from the linked persona, so // persona harness edits propagate on the next spawn; an explicit per-agent - // override wins. `agent_args` and `mcp_command` are pure derivations of the - // command, so we recompute them from the effective value rather than the - // frozen record snapshot. Mirrors the model resolution below. + // override wins. `agent_args` is derived from the command. An explicit + // record-level `mcp_command` is also an override; otherwise runtime metadata + // supplies the default MCP sidecar. Mirrors the summary path above. let personas = super::load_personas(app).unwrap_or_default(); let teams = super::load_teams(app).unwrap_or_default(); // Load global config once; used for runtime_metadata_env_vars (model/provider fallback) @@ -475,9 +485,7 @@ pub fn spawn_agent_child( .map_err(|error| format!("failed to clone log handle: {error}"))?; let resolved_acp_command = resolve_command(&record.acp_command) .ok_or_else(|| missing_command_message(&record.acp_command, "ACP harness command"))?; - let effective_mcp_command = known_acp_runtime(effective_command) - .and_then(|r| r.mcp_command) - .unwrap_or(""); + let effective_mcp_command = effective_mcp_command(record, effective_command); let resolved_mcp_command: Option = if effective_mcp_command.is_empty() { None } else { diff --git a/desktop/src-tauri/src/managed_agents/runtime/process.rs b/desktop/src-tauri/src/managed_agents/runtime/process.rs index 37eb5659a4..477f767c0c 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/process.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/process.rs @@ -22,6 +22,8 @@ pub(crate) const KNOWN_AGENT_BINARIES: &[&str] = &[ // invocations — not listed here. "buzz-dev-mcp", "buzz_dev_mcp", + "buzz-message-mcp", + "buzz_message_mcp", ]; /// Script interpreters that may host managed agent wrappers (e.g. npm shims). diff --git a/desktop/src-tauri/src/managed_agents/runtime/tests.rs b/desktop/src-tauri/src/managed_agents/runtime/tests.rs index 762b0fe2a6..201e2e0e94 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/tests.rs @@ -115,6 +115,18 @@ fn unknown_command_returns_none() { assert!(known_acp_runtime("custom-agent").is_none()); } +#[test] +fn explicit_record_mcp_command_overrides_runtime_metadata() { + let mut record = fixture(RespondTo::Anyone, vec![], None); + record.agent_command = "hermes-acp".into(); + record.mcp_command = "/opt/buzz-message-mcp".into(); + + assert_eq!( + super::effective_mcp_command(&record, "hermes-acp"), + "/opt/buzz-message-mcp" + ); +} + // ── build_respond_to_env tests ─────────────────────────────────────── use super::test_fixtures::{expected_mode, expected_owner_only, fixture}; diff --git a/desktop/src-tauri/tauri.conf.json b/desktop/src-tauri/tauri.conf.json index 006d478181..63b651ab84 100644 --- a/desktop/src-tauri/tauri.conf.json +++ b/desktop/src-tauri/tauri.conf.json @@ -57,6 +57,7 @@ "binaries/buzz-agent", "binaries/buzz-backend-kubernetes", "binaries/buzz-dev-mcp", + "binaries/buzz-message-mcp", "binaries/git-credential-nostr", "binaries/buzz" ], diff --git a/scripts/bundle-sidecars.sh b/scripts/bundle-sidecars.sh index 8ea5fe2bbe..7db6b740c0 100755 --- a/scripts/bundle-sidecars.sh +++ b/scripts/bundle-sidecars.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash set -euo pipefail -SIDECARS=(buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz) +SIDECARS=(buzz-acp buzz-agent buzz-dev-mcp buzz-message-mcp git-credential-nostr buzz) HOST=$(rustc -vV | sed -n 's|host: ||p') TARGET=${1:-$HOST} if [[ "$TARGET" != *windows* ]]; then