diff --git a/Cargo.lock b/Cargo.lock index 5c79f7eebc..65c3e90467 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1242,7 +1242,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", @@ -5562,7 +5562,7 @@ dependencies = [ "rand 0.10.1", "regex-lite", "reqwest 0.13.4", - "rmcp 3.2.0", + "rmcp 3.4.1", "rpassword", "rustls", "same-file", @@ -5674,7 +5674,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", @@ -7664,6 +7664,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" @@ -8573,14 +8587,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", @@ -8592,9 +8602,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", @@ -8607,10 +8617,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", @@ -8626,22 +8636,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-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), } } diff --git a/crates/buzz-agent/tests/regressions.rs b/crates/buzz-agent/tests/regressions.rs index fd5042a116..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")) @@ -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}" 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