mirror of
https://github.com/block/buzz.git
synced 2026-10-27 20:41:57 +00:00
Merge remote-tracking branch 'origin/main' into pinky/windows-oauth-private-file
Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
This commit is contained in:
Brain
51
Cargo.lock
generated
51
Cargo.lock
generated
@@ -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",
|
||||
|
||||
2
Justfile
2
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
|
||||
|
||||
@@ -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 `<context>`, 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 `<context>`.
|
||||
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<SteerAckEvent>,
|
||||
) -> 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 `<context>` 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::<pool::SteerAck>();
|
||||
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<pool::SteerRequest>, Option<ControlSignal>) {
|
||||
let channel_id = Uuid::new_v4();
|
||||
let ingress =
|
||||
|event: nostr::Event, edit: Option<queue::ResolvedEdit>| 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 `<context>` 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
|
||||
|
||||
@@ -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<SessionScope, Instant>,
|
||||
/// Number of events in each in-flight batch (for expiry logging).
|
||||
in_flight_batch_sizes: HashMap<SessionScope, usize>,
|
||||
/// Reply thread of each in-flight turn: the [`reply_thread`] of the batch
|
||||
/// event whose `<context>` routes the turn's replies (its last event).
|
||||
in_flight_reply_threads: HashMap<SessionScope, String>,
|
||||
retry_after: HashMap<SessionScope, Instant>,
|
||||
/// Per-scope retry attempt counter for exponential backoff / dead-lettering.
|
||||
retry_counts: HashMap<SessionScope, u32>,
|
||||
@@ -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
|
||||
/// `<context>`, so the turn keeps replying where its own `<context>`
|
||||
/// 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 `<context>`.
|
||||
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 {
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String> = 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}"
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<CallToolResult, ErrorData> {
|
||||
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<CallToolResult, ErrorData> {
|
||||
Ok(CallToolResult::error(vec![Content::text(s)]))
|
||||
Ok(CallToolResult::error(vec![ContentBlock::text(s)]))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -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<CallToolResu
|
||||
);
|
||||
|
||||
Ok(CallToolResult::success(vec![
|
||||
Content::text(header),
|
||||
Content::image(encoded, prepared.mime.to_string()),
|
||||
ContentBlock::text(header),
|
||||
ContentBlock::image(encoded, prepared.mime.to_string()),
|
||||
]))
|
||||
}
|
||||
|
||||
|
||||
150
crates/buzz-dev-mcp/tests/lifecycle.rs
Normal file
150
crates/buzz-dev-mcp/tests/lifecycle.rs
Normal file
@@ -0,0 +1,150 @@
|
||||
//! Exercise both MCP lifecycles against the shipped stdio server, without an LLM.
|
||||
use serde_json::{json, Value};
|
||||
use std::process::Stdio;
|
||||
use std::time::Duration;
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
|
||||
use tokio::process::{Child, ChildStdin, ChildStdout, Command};
|
||||
use tokio::time::timeout;
|
||||
|
||||
struct Server {
|
||||
child: Child,
|
||||
input: ChildStdin,
|
||||
output: Lines<BufReader<ChildStdout>>,
|
||||
work: tempfile::TempDir,
|
||||
meta: Option<Value>,
|
||||
}
|
||||
|
||||
impl Server {
|
||||
fn start(meta: Option<Value>) -> 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;
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user