From 2c270cf9d7849b6fc596a17608d78d91e2c10507 Mon Sep 17 00:00:00 2001 From: Alex Rosenzweig <64241648+shellz-n-stuff@users.noreply.github.com> Date: Fri, 2 Oct 2026 11:10:33 -0500 Subject: [PATCH 1/3] test(agent): synchronize handoff steering with tool approval (#8042) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary `handoff_cap_binds_within_a_single_turn` can finish the turn before its steer is accepted. It also requests `test_tool`, which the fake MCP server does not expose, so the supposed tool round fails immediately. Call the real `cap_test__tool_0` fixture and hold its permission response until the steer is acknowledged. Assert that the next model request contains the steer and the handoff cap still prevents a second summary. This changes one regression test; production code is unchanged. ## Validation - Reproduced the exact CI failure on unchanged `main` with a temporary 250 ms reader delay; the repaired test passed with the same delay. The delay is not committed. - Temporarily reset the handoff counter when a steer is queued: the repaired test failed. Restored production code afterward. - All six handoff regressions passed: `cargo test -p buzz-agent --locked --test regressions handoff_`. - `just ci` passed on the combined stack: repository checks, Rust and Tauri tests/lint, desktop tests/build, web build, and mobile checks/tests. - All 54 native-agent regressions passed on the combined stack, including this test. ## Stack 1. **This PR → `main`**: deterministic handoff regression. 2. **#8037 → this branch**: Goose MCP discovery compatibility. Merge this PR first, then retarget #8037 to `main`. Signed-off-by: Alex Rosenzweig --- crates/buzz-agent/tests/regressions.rs | 113 +++++++++++-------------- 1 file changed, 48 insertions(+), 65 deletions(-) diff --git a/crates/buzz-agent/tests/regressions.rs b/crates/buzz-agent/tests/regressions.rs index fd5042a116..d47aec6c3a 100644 --- a/crates/buzz-agent/tests/regressions.rs +++ b/crates/buzz-agent/tests/regressions.rs @@ -3338,16 +3338,16 @@ async fn handoff_cap_binds_within_a_single_turn() { // req 1: turn 1 complete() → usage=950 (over threshold=900) // req 2: turn 2 round 0 summarize() → summary (handoff_attempts: 0→1) // req 3: turn 2 round 0 complete() → tool_call + usage=950 (re-arms gate) - // [fake-mcp tool executes; steer queued while run is active] - // req 4: turn 2 round 1 preflight → 950 >= 900 AND attempts=1 >= max=1 + // [steer accepted while fake-mcp tool waits for approval] + // turn 2 round 1 preflight → 950 >= 900 AND attempts=1 >= max=1 // → WARN, skip (cap exhausted for this turn) - // req 5: turn 2 round 1 complete() → end_turn (steer text folded into messages) + // req 4: turn 2 round 1 complete() → end_turn (steer text folded into messages) let fake_mcp = env!("CARGO_BIN_EXE_fake-mcp"); // Build a tool-call response that also carries usage so the gate re-arms // on round 1's preflight (without usage, last_request_input_tokens is None // after the handoff clears it, and the byte-fallback won't fire on tiny history). let tool_call_with_usage = { - let mut v = openai_tool_call("tc-1", "test_tool", json!({})); + let mut v = openai_tool_call("tc-1", "cap_test__tool_0", json!({})); v["usage"] = json!({ "prompt_tokens": 950u64, "completion_tokens": 5, @@ -3377,7 +3377,7 @@ async fn handoff_cap_binds_within_a_single_turn() { ) .await; - // Init with the fake MCP server so test_tool is available. + // Init with the fake MCP server so cap_test__tool_0 is available. h.send( "initialize", json!({"protocolVersion":1,"clientCapabilities":{}}), @@ -3421,76 +3421,59 @@ async fn handoff_cap_binds_within_a_single_turn() { ) .await; - // Drain until the final response, approving tool-permission requests, - // capturing the activeRunId once it is broadcast, sending one steer, - // and verifying that it is accepted in the live run. - let mut run_id: Option = None; - let mut steer_id: i64 = -1; - let mut steer_accepted = false; - loop { - let v = h.recv().await; - - // Capture the run id from the first session/update that carries it, - // then immediately queue a steer. This must happen before round 1 so - // the steer text is present but the cap check still fires — proving - // the counter is not reset by the steer path. - if run_id.is_none() { - if let Some(rid) = v["params"]["update"]["_meta"]["goose"]["activeRunId"].as_str() { - run_id = Some(rid.to_owned()); - steer_id = h - .send( - "_goose/unstable/session/steer", - json!({ - "sessionId": sid, - "expectedRunId": rid, - "prompt": [{"type":"text","text":"STEER-CANARY: also consider the edge case"}], - }), - ) - .await; - } - } - - // Steer response: assert it was accepted in the live run. - if steer_id >= 0 && v["id"] == json!(steer_id) { - assert!( - v.get("result").is_some(), - "steer must be accepted while the run is active; got: {v}" - ); - assert_eq!( - v["result"]["runId"].as_str(), - run_id.as_deref(), - "steer must reference the live run id" - ); - steer_accepted = true; - continue; - } - - if v.get("method") == Some(&json!("session/request_permission")) { - h.write(approve_permission(&v)).await; - continue; - } - if v["id"] == json!(p2) { - assert!( - v.get("result").is_some(), - "turn 2 must succeed even when cap blocks round-1 handoff; got: {v}" - ); - break; - } - } + let update = h + .recv_until(|v| v["params"]["update"]["_meta"]["goose"]["activeRunId"].is_string()) + .await; + let run_id = update["params"]["update"]["_meta"]["goose"]["activeRunId"] + .as_str() + .unwrap(); + // Hold tool approval after the first handoff so the steer is accepted + // before round 1, regardless of how the agent and test are scheduled. + let permission = h + .recv_until(|v| v["method"] == "session/request_permission" || v["id"] == json!(p2)) + .await; + assert_eq!(permission["method"], "session/request_permission"); + let steer_id = h + .send( + "_goose/unstable/session/steer", + json!({ + "sessionId": sid, + "expectedRunId": run_id, + "prompt": [{"type":"text","text":"STEER-CANARY: also consider the edge case"}], + }), + ) + .await; + let steer = h.recv_until(|v| v["id"] == json!(steer_id)).await; + assert_eq!( + steer["result"]["runId"].as_str(), + Some(run_id), + "steer must be accepted in the live run; got: {steer}" + ); + h.write(approve_permission(&permission)).await; + let response = h.recv_until(|v| v["id"] == json!(p2)).await; assert!( - steer_accepted, - "steer was never accepted during turn 2; the steer arm is missing coverage" + response.get("result").is_some(), + "turn 2 must succeed even when cap blocks round-1 handoff; got: {response}" ); // 4 LLM requests: seed + summarize + tool-call-with-usage + final-complete. - let count = llm.captured.lock().await.len(); + let requests = llm.captured.lock().await; + let count = requests.len(); assert_eq!( count, 4, "expected 4 LLM requests (seed + summarize + tool-call + final); got {count}" ); - let stderr = h.stderr_text(); + assert!( + requests[3]["messages"].to_string().contains("STEER-CANARY"), + "the final request must include the accepted steer" + ); + drop(requests); + + let stderr = h + .wait_for_stderr("handoff cap reached", Duration::from_secs(5)) + .await; assert!( stderr.contains("handoff cap reached"), "expected cap-reached WARN in stderr; got: {stderr}" From 4f51b9e1010e086a16c099cd8d8218ca974a5e18 Mon Sep 17 00:00:00 2001 From: Alex Rosenzweig <64241648+shellz-n-stuff@users.noreply.github.com> Date: Fri, 2 Oct 2026 11:24:25 -0500 Subject: [PATCH 2/3] fix(mcp): support Goose discovery handshake (#8037) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Goose 1.52 starts MCP connections with `server/discover`. `buzz-dev-mcp` rejects that request with `ExpectedInitializeRequest`, preventing Goose from loading its Buzz tools. Upgrade the tool server to rmcp 3.4.1 and adopt the renamed SDK types. Subprocess tests exercise discovery and legacy initialization, tool listing, file reads/edits, invalid arguments, connection survival, and EOF shutdown. Include the server in both Rust unit-test runners and make the cancellation regression approve its shell call before testing cancellation. The native agent retains its independent rmcp 1.x client. Cargo also unifies the existing 3.x dependency used by the mesh smoke harness. ## Validation - Normal `git push` passed all hooks, including Rust tests, Tauri checks, and repository policy checks. - `just ci` passed on the final stack: repository checks, Rust and Tauri tests/lint, desktop tests/build, web build, and mobile checks/tests. - MCP: 78 existing tests and two subprocess lifecycle tests passed; the discovery regression fails with the old SDK. - All 54 native-agent regressions passed on the final stack, including cancellation of a real MCP shell process. - Previously exercised unmodified Goose 1.52 and 1.47 with Sandpit and the real tool server. Allowed workspace operations and permitted network traffic succeeded; disallowed file and network operations were denied. - Live staging desktop check with Goose 1.52: the reply appeared in Buzz, an allowed request returned HTTP 200, and a disallowed request returned CONNECT 403. ## Stack 1. **#8042 → `main`**: deterministic handoff regression. 2. **This PR → #8042**: Goose MCP discovery compatibility. Merge the lower PR first, then retarget this PR to `main`. This diff contains only the compatibility change and its tests; the relay build repair is already on `main` in #8036. --------- Signed-off-by: Alex Rosenzweig --- Cargo.lock | 51 ++++----- Justfile | 2 + crates/buzz-agent/tests/regressions.rs | 4 +- crates/buzz-dev-mcp/Cargo.toml | 4 +- crates/buzz-dev-mcp/src/lib.rs | 6 +- crates/buzz-dev-mcp/src/shell.rs | 10 +- crates/buzz-dev-mcp/src/todo.rs | 6 +- crates/buzz-dev-mcp/src/view_image.rs | 8 +- crates/buzz-dev-mcp/tests/lifecycle.rs | 150 +++++++++++++++++++++++++ scripts/run-tests.sh | 4 + 10 files changed, 200 insertions(+), 45 deletions(-) create mode 100644 crates/buzz-dev-mcp/tests/lifecycle.rs diff --git a/Cargo.lock b/Cargo.lock index 0e100811c6..dd15b2c222 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1241,7 +1241,7 @@ dependencies = [ "nix 0.31.3", "nostr 0.44.7", "reqwest 0.13.4", - "rmcp 1.8.0", + "rmcp 3.4.1", "rustls", "schemars 1.2.1", "serde", @@ -5553,7 +5553,7 @@ dependencies = [ "rand 0.10.1", "regex-lite", "reqwest 0.13.4", - "rmcp 3.2.0", + "rmcp 3.4.1", "rpassword", "rustls", "same-file", @@ -5665,7 +5665,7 @@ dependencies = [ "prost 0.14.3", "prost-build 0.14.3", "protoc-bin-vendored", - "rmcp 3.2.0", + "rmcp 3.4.1", "schemars 1.2.1", "serde", "serde_json", @@ -7655,6 +7655,20 @@ dependencies = [ "windows 0.62.2", ] +[[package]] +name = "process-wrap" +version = "10.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f21b97672d2dc848e7b25701ab4535618b92f4861c13cc3f7f7bed52ad3c8da" +dependencies = [ + "futures", + "indexmap 2.14.0", + "nix 0.31.3", + "tokio", + "tracing", + "windows 0.62.2", +] + [[package]] name = "proptest" version = "1.11.0" @@ -8564,14 +8578,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d1f571c72940a19d9532fe52dbea8bc9912bf1d766c2970bb824056b86f3f59" dependencies = [ "async-trait", - "base64 0.22.1", "chrono", "futures", - "pastey", "pin-project-lite", - "process-wrap", - "rmcp-macros 1.8.0", - "schemars 1.2.1", + "process-wrap 9.1.0", "serde", "serde_json", "thiserror 2.0.18", @@ -8583,9 +8593,9 @@ dependencies = [ [[package]] name = "rmcp" -version = "3.2.0" +version = "3.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42b6914fac0be956fe704a38239c3f44a9f841d1b06a5713d2f638065593f5b5" +checksum = "5b6317cd8c13e3ec9033cf2aa5aa92cd743f0f4f8a93cddc42ea0ae6ce3b8898" dependencies = [ "async-trait", "base64 0.23.1", @@ -8598,10 +8608,10 @@ dependencies = [ "indexmap 2.14.0", "pastey", "pin-project-lite", - "process-wrap", + "process-wrap 10.0.1", "rand 0.10.1", "reqwest 0.13.4", - "rmcp-macros 3.2.0", + "rmcp-macros", "schemars 1.2.1", "serde", "serde_json", @@ -8617,22 +8627,9 @@ dependencies = [ [[package]] name = "rmcp-macros" -version = "1.8.0" +version = "3.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1aad0035b69380782d78ea95b508327e6deaa2235909053e596eea8f27b5e1d5" -dependencies = [ - "darling 0.23.0", - "proc-macro2", - "quote", - "serde_json", - "syn 2.0.117", -] - -[[package]] -name = "rmcp-macros" -version = "3.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdf1c49bd4d52014b94db0877410db273c2008f01628b0252a2e9460ad9b7fda" +checksum = "d2ee366004e5f87c103be91c6c09337811517ae8d2f919ece9da7c21444b85c5" dependencies = [ "darling 0.24.1", "proc-macro2", diff --git a/Justfile b/Justfile index 01afccffc5..dc02757a02 100644 --- a/Justfile +++ b/Justfile @@ -387,6 +387,8 @@ test-unit: # buzz-acp owns the relay-to-agent trust boundary. Run its tests here so # forged relay events cannot regain a path into agent routing unnoticed. cargo nextest run -p buzz-acp + # Exercise the real MCP server for modern and legacy clients. + cargo nextest run -p buzz-dev-mcp # buzz-db migrator/lint tests: pure SQL-parsing unit tests (no infra). # They guard the embedded-migrator invariant (the complete checked-in # additive migration set; legacy cutover/backfill remains an operator diff --git a/crates/buzz-agent/tests/regressions.rs b/crates/buzz-agent/tests/regressions.rs index d47aec6c3a..a6feb989b6 100644 --- a/crates/buzz-agent/tests/regressions.rs +++ b/crates/buzz-agent/tests/regressions.rs @@ -1444,8 +1444,8 @@ async fn cancel_kills_inflight_tool_via_mcp_notification() { ) .await; - // Wait for the tool call to be in-progress. - h.recv_until(|v| { + // Approve the tool so this test reaches execution before cancelling it. + h.recv_until_approving(|v| { v.get("params") .and_then(|p| p.get("update")) .and_then(|u| u.get("status")) diff --git a/crates/buzz-dev-mcp/Cargo.toml b/crates/buzz-dev-mcp/Cargo.toml index 08d1ffabe3..5b3d7a9c9d 100644 --- a/crates/buzz-dev-mcp/Cargo.toml +++ b/crates/buzz-dev-mcp/Cargo.toml @@ -20,7 +20,9 @@ tokio = { workspace = true } tokio-util = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -rmcp = { workspace = true } +# Discovery-based clients (including Goose 1.52) require the modern lifecycle. +# Keep the native agent's independent MCP client dependency unchanged. +rmcp = { version = "3.4.1", features = ["server", "transport-io", "macros"] } schemars = { workspace = true } similar = "3" tempfile = "3" diff --git a/crates/buzz-dev-mcp/src/lib.rs b/crates/buzz-dev-mcp/src/lib.rs index 837ec292ab..00f48dccbd 100644 --- a/crates/buzz-dev-mcp/src/lib.rs +++ b/crates/buzz-dev-mcp/src/lib.rs @@ -2,7 +2,7 @@ #![cfg_attr(windows, deny(unsafe_code))] use rmcp::{ handler::server::{router::tool::ToolRouter, wrapper::Parameters}, - model::{CallToolResult, ServerCapabilities, ServerInfo}, + model::{CallToolResult, ServerCapabilities, ServerConfig}, tool, tool_handler, tool_router, transport::stdio, ErrorData, ServerHandler, ServiceExt, @@ -125,8 +125,8 @@ impl DevMcp { #[tool_handler(router = self.tool_router)] impl ServerHandler for DevMcp { - fn get_info(&self) -> ServerInfo { - ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + fn get_info(&self) -> ServerConfig { + ServerConfig::new(ServerCapabilities::builder().enable_tools().build()) .with_server_info(rmcp::model::Implementation::new( "buzz-dev-mcp", env!("CARGO_PKG_VERSION"), diff --git a/crates/buzz-dev-mcp/src/shell.rs b/crates/buzz-dev-mcp/src/shell.rs index 13e350689d..028e8ec8da 100644 --- a/crates/buzz-dev-mcp/src/shell.rs +++ b/crates/buzz-dev-mcp/src/shell.rs @@ -1,5 +1,5 @@ use crate::shim::Shim; -use rmcp::model::{CallToolResult, Content}; +use rmcp::model::{CallToolResult, ContentBlock}; use rmcp::ErrorData; use schemars::JsonSchema; use serde::Deserialize; @@ -161,7 +161,7 @@ pub async fn run( let bash = match &state.resolved_shell { Ok((path, _)) => path.clone(), - Err(msg) => return Ok(CallToolResult::error(vec![Content::text(msg.clone())])), + Err(msg) => return Ok(CallToolResult::error(vec![ContentBlock::text(msg.clone())])), }; let shell_arg = shell_flag(&bash); let mut cmd = Command::new(&bash); @@ -179,7 +179,7 @@ pub async fn run( let mut child = match cmd.spawn() { Ok(c) => c, Err(e) => { - return Ok(CallToolResult::error(vec![Content::text(format!( + return Ok(CallToolResult::error(vec![ContentBlock::text(format!( "failed to spawn shell: {e}" ))])); } @@ -230,7 +230,7 @@ pub async fn run( } stdout_handle.abort(); stderr_handle.abort(); - return Ok(CallToolResult::error(vec![Content::text("cancelled")])); + return Ok(CallToolResult::error(vec![ContentBlock::text("cancelled")])); } r = tokio::time::timeout(timeout_dur, child.wait()) => match r { Ok(Ok(s)) => (Some(s), false), @@ -316,7 +316,7 @@ pub async fn run( }); let text = serde_json::to_string_pretty(&body).unwrap_or_else(|_| "{}".into()); kill_group.disarm(); - Ok(CallToolResult::success(vec![Content::text(text)])) + Ok(CallToolResult::success(vec![ContentBlock::text(text)])) } /// The flag used to pass a command string to the shell. diff --git a/crates/buzz-dev-mcp/src/todo.rs b/crates/buzz-dev-mcp/src/todo.rs index de218356f2..6ab6ab2d8e 100644 --- a/crates/buzz-dev-mcp/src/todo.rs +++ b/crates/buzz-dev-mcp/src/todo.rs @@ -11,7 +11,7 @@ //! On replacement, open items that disappear without being marked done //! trigger a soft warning appended to the tool response. -use rmcp::model::{CallToolResult, Content}; +use rmcp::model::{CallToolResult, ContentBlock}; use rmcp::ErrorData; use schemars::JsonSchema; use serde::Deserialize; @@ -239,12 +239,12 @@ fn render_items(items: &[Item]) -> String { /// Wrap a string result as an MCP CallToolResult with text content. pub fn text_result(s: String) -> Result { - Ok(CallToolResult::success(vec![Content::text(s)])) + Ok(CallToolResult::success(vec![ContentBlock::text(s)])) } /// Wrap an error string as an MCP CallToolResult with isError=true. pub fn error_result(s: String) -> Result { - Ok(CallToolResult::error(vec![Content::text(s)])) + Ok(CallToolResult::error(vec![ContentBlock::text(s)])) } #[cfg(test)] diff --git a/crates/buzz-dev-mcp/src/view_image.rs b/crates/buzz-dev-mcp/src/view_image.rs index 8f8ea0ede7..f1f6978818 100644 --- a/crates/buzz-dev-mcp/src/view_image.rs +++ b/crates/buzz-dev-mcp/src/view_image.rs @@ -6,7 +6,7 @@ //! Design goals: tiny surface, no protocol-specific branching, and a //! "reasonable resolution" that fits comfortably inside both Anthropic's //! recommended ≤1568px / ≤5 MiB image budget and OpenAI's high-detail tile -//! size sweet spot. The MCP host translates `Content::image(data, mime)` +//! size sweet spot. The MCP host translates `ContentBlock::image(data, mime)` //! into the right provider-native shape on our behalf (see Goose's //! `providers::utils::convert_image` for a reference implementation). @@ -18,7 +18,7 @@ use image::{ DynamicImage, ExtendedColorType, ImageEncoder, ImageReader, Limits, }; use rmcp::{ - model::{CallToolResult, Content}, + model::{CallToolResult, ContentBlock}, ErrorData, }; use schemars::JsonSchema; @@ -101,8 +101,8 @@ pub async fn run(state: &SharedState, p: ViewImageParams) -> Result>, + work: tempfile::TempDir, + meta: Option, +} + +impl Server { + fn start(meta: Option) -> Self { + let work = tempfile::tempdir().unwrap(); + std::fs::write(work.path().join("fixture.txt"), "sunflower\n").unwrap(); + let mut child = Command::new(env!("CARGO_BIN_EXE_buzz-dev-mcp")) + .current_dir(work.path()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .kill_on_drop(true) + .spawn() + .unwrap(); + Self { + input: child.stdin.take().unwrap(), + output: BufReader::new(child.stdout.take().unwrap()).lines(), + child, + work, + meta, + } + } + + async fn request(&mut self, id: u32, method: &str, mut params: Value) -> Value { + if let Some(meta) = &self.meta { + params["_meta"] = meta.clone(); + } + let frame = json!({"jsonrpc":"2.0", "id":id, "method":method, "params":params}); + self.input + .write_all(format!("{frame}\n").as_bytes()) + .await + .unwrap(); + let line = timeout(Duration::from_secs(10), self.output.next_line()) + .await + .expect("server response deadline") + .unwrap() + .expect("server closed stdout"); + let response: Value = serde_json::from_str(&line).unwrap(); + assert_eq!(response["id"], id); + response + } + + async fn check_tools_and_shutdown(mut self) { + let listed = self.request(2, "tools/list", json!({})).await; + let tools = listed["result"]["tools"].as_array().unwrap(); + for name in ["shell", "read_file", "str_replace"] { + assert!(tools.iter().any(|tool| tool["name"] == name), "{listed}"); + } + let read = self + .request( + 3, + "tools/call", + json!({ + "name":"read_file", "arguments":{"path":"fixture.txt"} + }), + ) + .await; + assert!(read["result"]["content"][0]["text"] + .as_str() + .unwrap() + .contains("sunflower")); + let edit = self.request(4, "tools/call", json!({ + "name":"str_replace", "arguments":{"path":"fixture.txt", "old_str":"sunflower", "new_str":"orchid"} + })).await; + assert!(edit.get("error").is_none(), "{edit}"); + assert_eq!( + std::fs::read_to_string(self.work.path().join("fixture.txt")).unwrap(), + "orchid\n" + ); + let bad = self + .request( + 5, + "tools/call", + json!({ + "name":"read_file", "arguments":{} + }), + ) + .await; + assert_eq!(bad["result"]["isError"], true, "{bad}"); + // A rejected tool call must leave the connection usable. + let listed = self.request(6, "tools/list", json!({})).await; + assert!(listed["result"]["tools"].is_array(), "{listed}"); + drop(self.input); + let status = timeout(Duration::from_secs(10), self.child.wait()) + .await + .unwrap() + .unwrap(); + assert!(status.success(), "{status}"); + } +} + +#[tokio::test] +async fn legacy_initialize_preserves_tools() { + for version in ["2024-11-05", "2025-11-25"] { + let mut server = Server::start(None); + let reply = server + .request( + 1, + "initialize", + json!({ + "protocolVersion":version, "capabilities":{}, + "clientInfo":{"name":"legacy-client", "version":"1"} + }), + ) + .await; + assert_eq!(reply["result"]["protocolVersion"], version, "{reply}"); + server + .input + .write_all(b"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n") + .await + .unwrap(); + server.check_tools_and_shutdown().await; + } +} + +#[tokio::test] +async fn discover_preserves_tools() { + let mut server = Server::start(Some(json!({ + "io.modelcontextprotocol/protocolVersion":"2026-07-28", + "io.modelcontextprotocol/clientInfo":{"name":"discovery-client", "version":"1"}, + "io.modelcontextprotocol/clientCapabilities":{} + }))); + let reply = server.request(1, "server/discover", json!({})).await; + assert!( + reply["result"]["supportedVersions"] + .as_array() + .unwrap() + .contains(&json!("2026-07-28")), + "{reply}" + ); + assert_eq!( + reply["result"]["_meta"]["io.modelcontextprotocol/serverInfo"]["name"], + "buzz-dev-mcp" + ); + server.check_tools_and_shutdown().await; +} diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 524f60ed1b..96a16da9f2 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -110,6 +110,10 @@ run_unit_tests() { run_test_step "buzz-acp tests" \ cargo test -p buzz-acp -- --nocapture + # Keep MCP lifecycle coverage in step with the nextest path. + run_test_step "buzz-dev-mcp tests" \ + cargo test -p buzz-dev-mcp -- --nocapture + # buzz-db migrator/lint unit tests (no infra): guard the embedded-migrator # invariant (exactly the consolidated 0001; cutover/backfill stays an operator From 133fb9869228dab4c00cb2b9e259bf834a55dbf7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 2 Oct 2026 13:24:22 -0400 Subject: [PATCH 3/3] fix(acp): steer running turns with edited mentions (#6132) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🤖 ## Summary Some agents can take a new message into a turn that is already running. This is called "native steering." Before this PR, an edit that arrived during a turn always cancelled the turn and started it again. Now the edit is steered into the running turn, like any other new message. The steered edit still points at the original message and its thread, so the agent replies in the right place. - The steer text carries the edit's route: `Edit of: …`, plus the original's thread root. - **Steer only within the same reply thread.** A native steer keeps the running turn's ``, so the reply goes where that `` points. The queue now records each running turn's reply thread. A message is steered natively only when it replies in that same thread. Under channel sessions one session spans threads, so an edit for another thread (or another top-level message) takes the cancel-and-restart path. Its re-prompt carries the message's own full ``, with channel, project, and profile data. The same rule applies to ordinary steered messages, which had the same gap before this PR. - **Per-thread sessions:** the edit steers only a turn running in the same session as its original message. An edit in thread A is never added to a turn running in thread B. If no turn is running in that session, the edit waits in the queue as its own item. - **Channel sessions** (the default): the channel has one session. A running turn receives the edit natively only when the edit replies in that turn's thread. - If native steering is not available, or the transport fails or the agent rejects the steer, the existing cancel-and-restart fallback runs. If the agent returns another error, or the turn has already finished, the edit goes back to the queue for normal delivery. In every case the queued edit keeps its route. ## Details - #4741 looks up the original message once, before the edit is queued. This PR passes that stored route to the existing steer path, and uses the same session the edit was queued under. No new lookup or preparation step is needed. - Removal, re-add, and fallback all use the existing lifecycle for withheld steers. ### Stack **3 of 3**. #6131 is merged; this PR is now based on `main`. ### Related issue Part of #2540 (fixed by #6131). This PR adds the running-turn behavior. ### Testing - `cargo test -p buzz-acp --all-targets` passes. The one test that fails when the git setting `nostr.keyfile` is set is described in #4741. - Tests check that the steer text carries the original's route, and that the listener gives the steer path the same route and session it queues. - End-to-end tests send a routed edit through the listener's steer decision under channel sessions. An edit in the running turn's thread, and an edit of the running top-level message, steer natively with no cancel. An edit for thread B during a thread-A turn, and an edit of another top-level message, send no native steer and take cancel+merge. Removing the same-thread check makes both cross-thread tests fail. --------- Signed-off-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 266 ++++++++++++++++++++++++++++++----- crates/buzz-acp/src/queue.rs | 50 +++++++ crates/buzz-acp/src/scope.rs | 7 +- 3 files changed, 285 insertions(+), 38 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 3f74500d71..c621785f3b 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -632,8 +632,9 @@ struct QueuedNormalListenerEvent { /// Visible message that owns this event's lifecycle reactions: the /// original for an edit, otherwise the event itself. reaction_target_id: String, - event_for_steer: nostr::Event, - prompt_tag_for_steer: String, + /// The admitted event as a native steer would render it, including an + /// edit's resolved original-message routing. + steer_event: queue::BatchEvent, } impl QueuedNormalListenerEvent { @@ -662,17 +663,19 @@ impl QueuedNormalListenerEvent { let Some(signal) = mode_gate_signal(handling, &self.effective_author, owner) else { return; }; - // Edits take the universal cancel+merge path: a native steer body does - // not yet carry the edit's original-message routing, whereas a - // requeued edit is re-dispatched as its own routed batch. + // A native steer keeps the running turn's ``, so it may only + // carry a message that replies in the same thread. Under the channel + // policy one session spans threads; a message for another thread takes + // the cancel+merge path, whose re-prompt carries its own ``. + let same_reply_thread = queue.in_flight_reply_thread(&self.scope) + == Some(self.steer_event.reply_thread().as_str()); let native_attempted = matches!(signal, ControlSignal::Steer) - && queue::edit_target_id(&self.event_for_steer).is_none() + && same_reply_thread && try_native_steer( pool, queue, self.scope.clone(), - self.event_for_steer, - self.prompt_tag_for_steer, + self.steer_event, steer_ack_tx, ); if !native_attempted { @@ -719,14 +722,19 @@ impl NormalListenerIngress { edit, } = self; let reaction_target_id = queue::reaction_target_id(&buzz_event.event); - let event_for_steer = buzz_event.event.clone(); - let prompt_tag_for_steer = prompt_tag.clone(); + let received_at = std::time::Instant::now(); + let steer_event = queue::BatchEvent { + event: buzz_event.event.clone(), + prompt_tag: prompt_tag.clone(), + received_at, + edit: edit.clone(), + }; let channel_id = buzz_event.channel_id; let accepted = queue.push(QueuedEvent { channel_id, scope: session_scope.clone(), event: buzz_event.event, - received_at: std::time::Instant::now(), + received_at, prompt_tag, edit, }); @@ -735,8 +743,7 @@ impl NormalListenerIngress { scope: session_scope, effective_author, reaction_target_id, - event_for_steer, - prompt_tag_for_steer, + steer_event, } } } @@ -4326,8 +4333,7 @@ fn try_native_steer( pool: &mut AgentPool, queue: &mut EventQueue, scope: scope::SessionScope, - event: nostr::Event, - prompt_tag: String, + be: queue::BatchEvent, steer_ack_tx: &mpsc::UnboundedSender, ) -> bool { let channel_id = scope.channel_id(); @@ -4344,22 +4350,11 @@ fn try_native_steer( // channel context and the actor's profile in the original prompt, // duplicating it here would defeat the point of non-cancelling // steering (which is to inject only what's new). - let (tag, closing) = queue::native_steer_framing(); - let event_id_hex = event.id.to_hex(); - let be = queue::BatchEvent { - event, - prompt_tag: prompt_tag.clone(), - received_at: std::time::Instant::now(), - edit: None, - }; - let event_block = queue::format_event_block(channel_id, None, &be, None); - let new_message = prompt_framing::semantic_section(tag, ""); - let event_section = prompt_framing::semantic_section_with_attributes( - "buzz-event", - &[("type", prompt_tag.as_str())], - &event_block, - ); - let body = format!("{new_message}\n\n{event_section}\n\n{closing}"); + // The caller steers natively only a message in the running turn's reply + // thread, so the turn's own `` still routes the reply. An edit's + // block names its original (`Edit of:`) and the original's thread root. + let event_id_hex = be.event.id.to_hex(); + let body = native_steer_body(channel_id, &be); let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); let request = pool::SteerRequest { @@ -4426,6 +4421,19 @@ fn try_native_steer( } } +/// Render the prompt delta sent by [`try_native_steer`]. +fn native_steer_body(channel_id: Uuid, be: &queue::BatchEvent) -> String { + let (tag, closing) = queue::native_steer_framing(); + let event_block = queue::format_event_block(channel_id, None, be, None); + let new_message = prompt_framing::semantic_section(tag, ""); + let event_section = prompt_framing::semantic_section_with_attributes( + "buzz-event", + &[("type", be.prompt_tag.as_str())], + &event_block, + ); + format!("{new_message}\n\n{event_section}\n\n{closing}") +} + // ── try_native_steer fallback-log tests ─────────────────────────────────────── // // Regression for the production tracing event `try_native_steer`'s Err arm @@ -4594,8 +4602,12 @@ mod try_native_steer_fallback_log_tests { pool, &mut queue, busy_scope.clone(), - event, - "mention".into(), + queue::BatchEvent { + event, + prompt_tag: "mention".into(), + received_at: std::time::Instant::now(), + edit: None, + }, &steer_ack_tx, ) }); @@ -9949,6 +9961,194 @@ mod edit_mention_admission_tests { } } +#[cfg(test)] +mod edit_native_steer_tests { + use super::*; + use crate::edit_routing::test_support::{edit_event, message}; + + /// A steered edit renders its original's routing, not the edit's own + /// bare `e` tag, so the live turn replies in the original's thread. + #[test] + fn native_steer_body_carries_edit_original_routing() { + let root = "ab".repeat(32); + let original = message(Some(&root)); + let edit = edit_event(&original.id.to_hex(), &[]); + let be = queue::BatchEvent { + event: edit.clone(), + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + edit: Some(queue::ResolvedEdit { + target_event_id: original.id.to_hex(), + target_thread_tags: queue::parse_thread_tags(&original), + }), + }; + let body = native_steer_body(Uuid::new_v4(), &be); + assert!( + body.contains(&format!("Edit of: {}", original.id.to_hex())), + "{body}" + ); + assert!(body.contains(&format!("root={root}")), "{body}"); + } + + /// The listener hands the steer path the same resolved route it queues. + #[test] + fn listener_steer_event_keeps_resolved_edit() { + let original = message(None); + let edit = edit_event(&original.id.to_hex(), &[]); + let channel_id = Uuid::new_v4(); + let resolved = queue::ResolvedEdit { + target_event_id: original.id.to_hex(), + target_thread_tags: queue::parse_thread_tags(&original), + }; + let ingress = NormalListenerIngress { + buzz_event: relay::BuzzEvent { + connection_generation: 0, + channel_id, + event: edit, + }, + effective_author: "author".into(), + prompt_tag: "@mention".into(), + edit: Some(resolved.clone()), + }; + let scope = ingress.session_scope(scope::SessionPolicy::Channel, false); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let queued = ingress.push(&mut queue, scope); + assert_eq!(queued.steer_event.edit, Some(resolved)); + assert_eq!(queued.reaction_target_id, original.id.to_hex()); + } + + /// Drive a routed edit of `original` through the listener's steer decision + /// while a turn for `running_event` is in flight under the channel policy. + /// Returns the native steer request, if any, and the control signal sent + /// to the running turn, if any. + fn steer_edit_into_running_turn( + running_event: nostr::Event, + original: &nostr::Event, + ) -> (Option, Option) { + let channel_id = Uuid::new_v4(); + let ingress = + |event: nostr::Event, edit: Option| NormalListenerIngress { + buzz_event: relay::BuzzEvent { + connection_generation: 0, + channel_id, + event, + }, + effective_author: "author".into(), + prompt_tag: "@mention".into(), + edit, + }; + + let mut queue = EventQueue::new(config::DedupMode::Queue); + let running = ingress(running_event, None); + let scope = running.session_scope(scope::SessionPolicy::Channel, false); + running.push(&mut queue, scope.clone()); + queue.flush_next().expect("running turn"); + assert!(queue.is_scope_in_flight(&scope)); + + let mut pool = AgentPool::from_slots(vec![]); + let (control_tx, mut control_rx) = tokio::sync::oneshot::channel(); + let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1); + let task = pool.join_set.spawn(async {}); + pool.task_map_mut().insert( + task.id(), + pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + scope: Some(scope.clone()), + turn_id: "t".into(), + recoverable_batch: None, + control_tx: Some(control_tx), + steer_tx: Some(steer_tx), + successful_steer_deliveries: HashSet::new(), + }, + ); + + let edit = edit_event(&original.id.to_hex(), &[]); + let resolved = queue::ResolvedEdit { + target_event_id: original.id.to_hex(), + target_thread_tags: queue::parse_thread_tags(original), + }; + let edit_ingress = ingress(edit, Some(resolved)); + assert_eq!( + edit_ingress.session_scope(scope::SessionPolicy::Channel, false), + scope, + "channel policy: one session spans every thread" + ); + let (ack_tx, _ack_rx) = mpsc::unbounded_channel(); + edit_ingress + .push(&mut queue, scope.clone()) + .steer_or_interrupt( + MultipleEventHandling::Steer, + None, + &mut pool, + &mut queue, + &ack_tx, + ); + (steer_rx.try_recv().ok(), control_rx.try_recv().ok()) + } + + /// A routed edit whose original is in the running turn's thread is + /// steered natively, carrying the original's route, and the running turn + /// is not cancelled. + #[tokio::test] + async fn routed_edit_in_running_thread_steers_natively() { + let root = "ab".repeat(32); + let original = message(Some(&root)); + let (steer, control) = steer_edit_into_running_turn(message(Some(&root)), &original); + + let request = steer.expect("edit is sent as a native steer"); + let body = request.prompt_blocks.join("\n"); + assert!( + body.contains(&format!("Edit of: {}", original.id.to_hex())), + "{body}" + ); + assert!(body.contains(&format!("root={root}")), "{body}"); + assert_eq!( + control, None, + "native steer must not cancel the running turn" + ); + } + + /// Editing the top-level message a turn is working on steers that turn: + /// replies to the edit belong to the same new thread. + #[tokio::test] + async fn edit_of_running_top_level_trigger_steers_natively() { + let original = message(None); + let (steer, control) = steer_edit_into_running_turn(original.clone(), &original); + + assert!(steer.is_some(), "edit is sent as a native steer"); + assert_eq!( + control, None, + "native steer must not cancel the running turn" + ); + } + + /// Under the channel policy a turn started from thread A can be running + /// when an edit whose original is in thread B arrives. A native steer + /// would leave the turn's `` replying to A, so the edit takes the + /// cancel+merge path instead; its re-prompt routes replies to B. + #[tokio::test] + async fn routed_edit_for_another_thread_cancels_and_merges() { + let thread_b = "ab".repeat(32); + let original = message(Some(&thread_b)); + let (steer, control) = steer_edit_into_running_turn(message(None), &original); + + assert!(steer.is_none(), "no native steer into thread A's turn"); + assert_eq!(control, Some(ControlSignal::Steer)); + } + + /// The same rule covers a top-level original: replies to it open a thread + /// rooted at the original, not at the running turn's top-level trigger. + #[tokio::test] + async fn routed_edit_for_another_top_level_message_cancels_and_merges() { + let original = message(None); + let (steer, control) = steer_edit_into_running_turn(message(None), &original); + + assert!(steer.is_none(), "no native steer across top-level threads"); + assert_eq!(control, Some(ControlSignal::Steer)); + } +} + #[cfg(test)] mod error_outcome_emission_tests { //! Pins the policy that error-class outcomes surface to the activity feed diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index e82a62a04b..a5b4b9d3f3 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -149,6 +149,11 @@ impl BatchEvent { pub fn routing_thread_tags(&self) -> ThreadTags { routing_thread_tags(&self.event, self.edit.as_ref()) } + + /// See [`reply_thread`]. + pub fn reply_thread(&self) -> String { + reply_thread(&self.event, self.edit.as_ref()) + } } /// Why a batch's prior turn was cancelled — controls how `format_prompt` @@ -236,6 +241,9 @@ pub struct EventQueue { in_flight_deadlines: HashMap, /// Number of events in each in-flight batch (for expiry logging). in_flight_batch_sizes: HashMap, + /// Reply thread of each in-flight turn: the [`reply_thread`] of the batch + /// event whose `` routes the turn's replies (its last event). + in_flight_reply_threads: HashMap, retry_after: HashMap, /// Per-scope retry attempt counter for exponential backoff / dead-lettering. retry_counts: HashMap, @@ -277,6 +285,7 @@ impl EventQueue { in_flight_scopes: HashSet::new(), in_flight_deadlines: HashMap::new(), in_flight_batch_sizes: HashMap::new(), + in_flight_reply_threads: HashMap::new(), retry_after: HashMap::new(), retry_counts: HashMap::new(), dedup_mode, @@ -424,6 +433,7 @@ impl EventQueue { ); self.in_flight_scopes.remove(&scope); self.in_flight_deadlines.remove(&scope); + self.in_flight_reply_threads.remove(&scope); // Recover any withheld goose-native steer events for the expired // scope back to the queue front so normal dispatch delivers // them. Unlike the in-flight batch above (already delivered to a @@ -467,6 +477,7 @@ impl EventQueue { .insert(scope.clone(), now + self.in_flight_deadline); self.in_flight_batch_sizes .insert(scope.clone(), cancelled.len()); + self.record_in_flight_reply_thread(&scope, &cancelled); return Some(FlushBatch { channel_id: scope.channel_id(), scope, @@ -504,6 +515,7 @@ impl EventQueue { .insert(scope.clone(), now + self.in_flight_deadline); self.in_flight_batch_sizes .insert(scope.clone(), events.len()); + self.record_in_flight_reply_thread(&scope, &events); // Merge any cancelled events stored by requeue_as_cancelled(). let cancelled_events = self.cancelled_batches.remove(&scope).unwrap_or_default(); @@ -556,6 +568,7 @@ impl EventQueue { self.in_flight_scopes.remove(&scope); self.in_flight_deadlines.remove(&scope); self.in_flight_batch_sizes.remove(&scope); + self.in_flight_reply_threads.remove(&scope); let now = Instant::now(); match self.retry_after.get(&scope) { // Active throttle → scope was requeued; keep retry_counts intact. @@ -774,6 +787,7 @@ impl EventQueue { ); self.in_flight_scopes.remove(&scope); self.in_flight_deadlines.remove(&scope); + self.in_flight_reply_threads.remove(&scope); // Symmetric with the flush_next expiry block: recover withheld // goose-native steer events for the expired scope so they are // not permanently orphaned in the side table. @@ -929,6 +943,29 @@ impl EventQueue { self.in_flight_scopes.contains(&scope.into_scope()) } + /// The reply thread of the turn in flight for `scope`, if any. + /// + /// A native steer adds a message to a running turn without a new + /// ``, so the turn keeps replying where its own `` + /// points. A message whose [`reply_thread`] differs (possible under the + /// channel session policy) must not be steered natively; the cancel+merge + /// path re-dispatches it with its own full ``. + pub fn in_flight_reply_thread(&self, scope: &SessionScope) -> Option<&str> { + self.in_flight_reply_threads.get(scope).map(String::as_str) + } + + fn record_in_flight_reply_thread(&mut self, scope: &SessionScope, events: &[BatchEvent]) { + match events.last() { + Some(last) => { + self.in_flight_reply_threads + .insert(scope.clone(), last.reply_thread()); + } + None => { + self.in_flight_reply_threads.remove(scope); + } + } + } + /// Whether any scope currently has a turn in flight. pub fn has_in_flight(&self) -> bool { !self.in_flight_scopes.is_empty() @@ -1175,6 +1212,19 @@ pub(crate) fn reaction_target_id(event: &Event) -> String { edit_target_id(event).unwrap_or_else(|| event.id.to_hex()) } +/// The thread that replies to `event` belong to: its routed thread root, or +/// the routed event itself when it is top-level (a reply opens a thread +/// rooted there). Lowercase, so equivalent hex spellings compare equal. +/// +/// This is the thread-session key, and it decides whether a message may be +/// steered natively into a running turn (see [`EventQueue::in_flight_reply_thread`]). +pub(crate) fn reply_thread(event: &Event, edit: Option<&ResolvedEdit>) -> String { + routing_thread_tags(event, edit) + .root_event_id + .unwrap_or_else(|| reaction_target_id(event)) + .to_ascii_lowercase() +} + /// Thread tags that route replies for `event`. See /// [`BatchEvent::routing_thread_tags`]. pub(crate) fn routing_thread_tags(event: &Event, edit: Option<&ResolvedEdit>) -> ThreadTags { diff --git a/crates/buzz-acp/src/scope.rs b/crates/buzz-acp/src/scope.rs index 1ec23ebeb7..a0dd2176fb 100644 --- a/crates/buzz-acp/src/scope.rs +++ b/crates/buzz-acp/src/scope.rs @@ -23,7 +23,7 @@ use nostr::Event; use uuid::Uuid; -use crate::queue::{reaction_target_id, routing_thread_tags, ResolvedEdit}; +use crate::queue::{reply_thread, ResolvedEdit}; /// Operator policy controlling how ACP provider sessions are scoped. /// @@ -150,12 +150,9 @@ impl SessionScope { return Self::Conversation { channel_id }; } - let root_event_id = routing_thread_tags(event, edit) - .root_event_id - .unwrap_or_else(|| reaction_target_id(event)); Self::Thread { channel_id, - root_event_id: root_event_id.to_ascii_lowercase(), + root_event_id: reply_thread(event, edit), } }