diff --git a/CHANGELOG.md b/CHANGELOG.md index 1fd2329..3280d1f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,17 @@ ### Fixed +- rmcp server worker: a single inbound message that was not an `initialize` request could + permanently stop the server for every client. All Nostr clients are multiplexed through one + rmcp service, and rmcp's pre-service handshake accepts only an `initialize` request as its + first message; anything else (a `notifications/initialized`, any other typed notification, a + response, or an error response) terminated the service, which cancelled the worker and closed + the transport. Nothing recovered it: every later client hung. Reaching it needed no + authentication, no allowlisted key, and no honest client, and an honest client could trigger it + on itself whenever the relay delivered its `notifications/initialized` before its `initialize`, + which Nostr does not order. The worker now satisfies the handshake itself at startup, before it + drains any relay-sourced message, so no inbound message can reach the pre-service handshake. + - CEP-41 open-stream: a deferred final response (one held back while a stream was still open) went out with routing tags only, so it carried neither the CEP-35 discovery tags nor the CEP-8 effective-mode disclosure. A client whose first server @@ -51,6 +62,17 @@ ### Changed +- rmcp server worker: every client now receives the handler's declared `protocolVersion` in its + `InitializeResult`. Because the worker satisfies the pre-service handshake itself, no client's + `initialize` runs rmcp's version negotiation any more, so the first client is answered with the + declared version rather than an echo of the version it requested. Clients after the first + already behaved this way, so the change makes all clients uniform, in the less conformant + direction: the multiplexed adapter no longer negotiates the protocol version per client. A + server announcement is unaffected for a handler that does not override the version, and + reflects the declared version for one that does. The per-client synthetic bootstrap that used + to inject an `initialize` ahead of a new client's first request is removed, since the handshake + is now satisfied before any client is served. + - The server transport's two response paths (the normal path and the CEP-41 deferred stream path) now compose their outbound tags through a single shared function instead of each writing the same routing, discovery, disclosure and pricing policy diff --git a/Cargo.toml b/Cargo.toml index 29a0fd9..414b856 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,6 +97,10 @@ required-features = ["rmcp", "test-utils"] name = "payments_negotiation_e2e" required-features = ["test-utils"] +[[test]] +name = "rmcp_handshake_survival" +required-features = ["rmcp", "test-utils"] + [dev-dependencies] # `start_paused` deterministic-time tests (tokio's "full" does not include test-util). tokio = { version = "1", features = ["test-util"] } diff --git a/src/rmcp_transport/worker.rs b/src/rmcp_transport/worker.rs index bd939e0..68ae72c 100644 --- a/src/rmcp_transport/worker.rs +++ b/src/rmcp_transport/worker.rs @@ -12,7 +12,6 @@ use crate::transport::server::{ }; use rmcp::model::GetExtensions; use rmcp::transport::worker::{Worker, WorkerContext, WorkerQuitReason}; -use std::collections::HashSet; use super::convert::{ internal_to_rmcp_client_rx, internal_to_rmcp_server_rx, rmcp_client_tx_to_internal, @@ -46,27 +45,6 @@ fn synthetic_initialized_notification() -> JsonRpcMessage { }) } -fn should_inject_stateless_bootstrap( - initialized_clients: &HashSet, - client_pubkey: &str, - message: &JsonRpcMessage, -) -> bool { - if initialized_clients.contains(client_pubkey) { - return false; - } - - matches!(message, JsonRpcMessage::Request(req) if req.method != "initialize") -} - -fn is_synthetic_initialize_message(message: &JsonRpcMessage) -> bool { - matches!( - message, - JsonRpcMessage::Request(req) - if req.method == "initialize" - && req.id == serde_json::json!(STATELESS_SYNTHETIC_EVENT_ID) - ) -} - /// rmcp server worker wrapper for ContextVM Nostr server transport. /// /// Multiplexes all connected clients through a single rmcp service instance. @@ -133,8 +111,48 @@ impl Worker for NostrServerWorker { ) })?; + // Satisfy rmcp's pre-init handshake worker-locally, before any relay-sourced + // message can be drained. The handler channel has one producer (this task) and + // is FIFO, so a synthetic initialize + initialized sent here reaches the service + // ahead of every inbound event, for every peer and ordering. + let startup: std::result::Result<(), WorkerQuitReason> = async { + let init = synthetic_initialize_message(); + let rmcp_init = internal_to_rmcp_server_rx(&init).ok_or_else(|| { + WorkerQuitReason::fatal( + Self::Error::Validation( + "failed converting synthetic initialize request to rmcp format".to_string(), + ), + "converting synthetic initialize request", + ) + })?; + context.send_to_handler(rmcp_init).await?; + + let initialized = synthetic_initialized_notification(); + let rmcp_initialized = internal_to_rmcp_server_rx(&initialized).ok_or_else(|| { + WorkerQuitReason::fatal( + Self::Error::Validation( + "failed converting synthetic initialized notification to rmcp format" + .to_string(), + ), + "converting synthetic initialized notification", + ) + })?; + context.send_to_handler(rmcp_initialized).await + } + .await; + // Share the loop's close-on-exit path: a send failure here means the + // WorkerTransport was already dropped (e.g. a serve()-vs-timeout race). Without + // this close, the transport's own tasks (event loop, cleanup, relay + // subscription) leak for the process lifetime, because nothing else runs between + // `take_message_receiver()` and the loop. + if let Err(reason) = startup { + if let Err(e) = self.transport.close().await { + tracing::warn!(target: LOG_TARGET, error = %e, "Failed to close server transport cleanly"); + } + return Err(reason); + } + let cancellation_token = context.cancellation_token.clone(); - let mut initialized_clients = HashSet::new(); let quit_reason = loop { tokio::select! { @@ -154,58 +172,12 @@ impl Worker for NostrServerWorker { .. } = incoming; - let should_inject_bootstrap = should_inject_stateless_bootstrap( - &initialized_clients, - &client_pubkey, - &message, - ); - - if should_inject_bootstrap { - let synthetic_init = synthetic_initialize_message(); - let Some(rmcp_init) = internal_to_rmcp_server_rx(&synthetic_init) else { - break WorkerQuitReason::fatal( - Self::Error::Validation( - "failed converting synthetic initialize request to rmcp format".to_string(), - ), - "converting synthetic initialize request", - ); - }; - - if let Err(reason) = context.send_to_handler(rmcp_init).await { - break reason; - } - - let initialized = synthetic_initialized_notification(); - let Some(rmcp_initialized) = internal_to_rmcp_server_rx(&initialized) else { - break WorkerQuitReason::fatal( - Self::Error::Validation( - "failed converting synthetic initialized notification to rmcp format".to_string(), - ), - "converting synthetic initialized notification", - ); - }; - - if let Err(reason) = context.send_to_handler(rmcp_initialized).await { - break reason; - } - - initialized_clients.insert(client_pubkey.clone()); - } - - if matches!(&message, JsonRpcMessage::Request(req) if req.method == "initialize") - || matches!(&message, JsonRpcMessage::Notification(n) if n.method == "notifications/initialized") - { - initialized_clients.insert(client_pubkey.clone()); - } - - // Rewrite real wire requests to the Nostr event_id. - // Synthetic stateless bootstrap messages must retain their - // sentinel ID so their responses can be dropped before they - // ever touch transport correlation. - if !is_synthetic_initialize_message(&message) { - if let JsonRpcMessage::Request(ref mut req) = message { + // Rewrite wire requests to the Nostr event_id. Every message + // reaching this loop came off the relay; the worker's own + // synthetic handshake pair goes straight to the handler at + // startup and never traverses it. + if let JsonRpcMessage::Request(ref mut req) = message { req.id = serde_json::json!(event_id); - } } if let Some(mut rmcp_msg) = internal_to_rmcp_server_rx(&message) { @@ -484,40 +456,6 @@ mod tests { use super::*; use crate::core::types::JsonRpcResponse; - #[test] - fn test_should_inject_stateless_bootstrap_for_first_non_initialize_request() { - let initialized_clients = HashSet::new(); - let message = JsonRpcMessage::Request(JsonRpcRequest { - jsonrpc: "2.0".to_string(), - id: serde_json::json!(1), - method: "tools/list".to_string(), - params: Some(serde_json::json!({})), - }); - - assert!(should_inject_stateless_bootstrap( - &initialized_clients, - "client-a", - &message, - )); - } - - #[test] - fn test_should_not_inject_stateless_bootstrap_for_real_initialize() { - let initialized_clients = HashSet::new(); - let message = JsonRpcMessage::Request(JsonRpcRequest { - jsonrpc: "2.0".to_string(), - id: serde_json::json!(1), - method: "initialize".to_string(), - params: Some(serde_json::json!({})), - }); - - assert!(!should_inject_stateless_bootstrap( - &initialized_clients, - "client-a", - &message, - )); - } - #[test] fn test_synthetic_initialize_keeps_sentinel_id() { let message = synthetic_initialize_message(); @@ -579,13 +517,6 @@ mod tests { } } - #[test] - fn test_is_synthetic_initialize_message_detects_sentinel() { - assert!(is_synthetic_initialize_message( - &synthetic_initialize_message() - )); - } - #[test] fn test_announcement_sentinel_differs_from_stateless_sentinel() { assert_ne!(ANNOUNCEMENT_REQUEST_ID, STATELESS_SYNTHETIC_EVENT_ID); @@ -606,8 +537,9 @@ mod tests { if let JsonRpcMessage::Response(ref resp) = response { let event_id = resp.id.as_str().unwrap(); assert_eq!(event_id, ANNOUNCEMENT_REQUEST_ID); - // Must not be confused with the stateless synthetic sentinel - assert!(!is_synthetic_initialize_message(&response)); + // Must not be confused with the stateless synthetic sentinel, whose + // response takes the drop path in `forward_server_internal`. + assert_ne!(event_id, STATELESS_SYNTHETIC_EVENT_ID); } } } diff --git a/tests/rmcp_handshake_survival.rs b/tests/rmcp_handshake_survival.rs new file mode 100644 index 0000000..ac68be1 --- /dev/null +++ b/tests/rmcp_handshake_survival.rs @@ -0,0 +1,511 @@ +//! Survival of the multiplexed rmcp server service against pre-initialize inbound traffic. +//! +//! rmcp's pre-service handshake accepts only an `initialize` request as its first message. The +//! worker multiplexes every Nostr client through one rmcp service, so a single inbound message of +//! any other shape reaching that handshake first terminates the service, cancels the worker, and +//! closes the transport for every client. These tests drive the real worker over `MockRelayPool` +//! and assert the server keeps serving. +//! +//! Timing uses the real tokio clock throughout (bounded `timeout` / `sleep`, never paused time): +//! the transport's own session and cleanup paths run on `std::time::Instant`. +//! +//! The server runs on `NostrServerTransportConfig::default()` (empty allowlist, not announced) +//! except where a test names the option it changes; hostile events are published in plaintext from +//! a separate linked pool, which the default `EncryptionMode::Optional` accepts. + +#![cfg(feature = "rmcp")] + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use contextvm_sdk::core::constants::{ + mcp_protocol_version, CTXVM_MESSAGES_KIND, SERVER_ANNOUNCEMENT_KIND, TOOLS_LIST_KIND, +}; +use contextvm_sdk::core::types::EncryptionMode; +use contextvm_sdk::relay::mock::MockRelayPool; +use contextvm_sdk::transport::base::BaseTransport; +use contextvm_sdk::transport::client::{NostrClientTransport, NostrClientTransportConfig}; +use contextvm_sdk::transport::server::{ + InboundContext, InboundMiddleware, Next, NostrServerTransport, NostrServerTransportConfig, +}; +use contextvm_sdk::{ + JsonRpcError, JsonRpcErrorResponse, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, + JsonRpcResponse, RelayPoolTrait, +}; +use nostr_sdk::prelude::*; +use rmcp::model::{ + ClientCapabilities, ClientInfo, Implementation, ProtocolVersion, ServerCapabilities, ServerInfo, +}; +use rmcp::{ClientHandler, ServerHandler, ServiceExt}; + +/// The sentinel id the worker's synthetic startup `initialize` carries. Its response must be +/// dropped before wire correlation, so this string must never appear in a published event. +const STATELESS_SENTINEL: &str = "contextvm-stateless-init"; + +// ── Handlers ────────────────────────────────────────────────────────────── + +/// Minimal MCP server, declaring rmcp's latest protocol version. `list_tools` is rmcp's default +/// (an empty list): these tests assert that a request is answered, not what it answers with. +struct SurvivalServer; + +impl ServerHandler for SurvivalServer { + fn get_info(&self) -> ServerInfo { + let mut info = ServerInfo::new(ServerCapabilities::builder().enable_tools().build()); + info.protocol_version = ProtocolVersion::LATEST; + info.server_info = Implementation::new("handshake-survival-server", "0.1.0"); + info + } +} + +/// Minimal MCP client. The requested protocol version varies per test, so it is a field. +struct SurvivalClient { + protocol_version: ProtocolVersion, +} + +impl SurvivalClient { + fn latest() -> Self { + Self::requesting(ProtocolVersion::LATEST) + } + + fn requesting(protocol_version: ProtocolVersion) -> Self { + Self { protocol_version } + } +} + +impl ClientHandler for SurvivalClient { + fn get_info(&self) -> ClientInfo { + ClientInfo::new( + ClientCapabilities::default(), + Implementation::new("handshake-survival-client", "0.1.0"), + ) + .with_protocol_version(self.protocol_version.clone()) + } +} + +/// A middleware that forwards everything. Registering any middleware at all moves inbound +/// messages off the event loop's inline path and onto a detached per-message task, which is the +/// ordering this exercises. +struct ForwardAll; + +#[async_trait] +impl InboundMiddleware for ForwardAll { + async fn handle(&self, message: JsonRpcMessage, _ctx: &InboundContext, next: Next) -> bool { + next.run(message).await + } +} + +// ── Helpers ─────────────────────────────────────────────────────────────── + +fn as_pool(pool: &Arc) -> Arc { + Arc::clone(pool) as Arc +} + +/// Give the spawned worker time to run `start()` and register its relay subscription, so a +/// hostile event published next is delivered live rather than missed. +async fn let_subscriptions_settle() { + tokio::time::sleep(Duration::from_millis(50)).await; +} + +fn request(id: &str, method: &str) -> JsonRpcMessage { + JsonRpcMessage::Request(JsonRpcRequest { + jsonrpc: "2.0".to_string(), + id: serde_json::json!(id), + method: method.to_string(), + params: Some(serde_json::json!({})), + }) +} + +fn initialize_request(id: &str) -> JsonRpcMessage { + JsonRpcMessage::Request(JsonRpcRequest { + jsonrpc: "2.0".to_string(), + id: serde_json::json!(id), + method: "initialize".to_string(), + params: Some(serde_json::json!({ + "protocolVersion": mcp_protocol_version(), + "capabilities": {}, + "clientInfo": { "name": "raw-wire-client", "version": "0.1.0" } + })), + }) +} + +fn initialized_notification() -> JsonRpcMessage { + JsonRpcMessage::Notification(JsonRpcNotification { + jsonrpc: "2.0".to_string(), + method: "notifications/initialized".to_string(), + params: None, + }) +} + +/// A plaintext ContextVM message event addressed to `server`, signed by `keys`. +fn wire_event(keys: &Keys, server: PublicKey, message: &JsonRpcMessage) -> Event { + let tags = BaseTransport::create_recipient_tags(&server); + contextvm_sdk::core::serializers::mcp_to_nostr_event(message, CTXVM_MESSAGES_KIND, tags) + .expect("serialize wire message") + .sign_with_keys(keys) + .expect("sign wire message") +} + +/// Publish one message onto the shared mock relay network from `keys`, which is not on the +/// server's allowlist (the default allowlist is empty, so nothing is). +async fn publish_from( + pool: &Arc, + keys: &Keys, + server: PublicKey, + message: &JsonRpcMessage, +) { + pool.publish_event(&wire_event(keys, server, message)) + .await + .expect("publish wire event"); +} + +async fn server_transport(pool: &Arc) -> NostrServerTransport { + NostrServerTransport::with_relay_pool(NostrServerTransportConfig::default(), as_pool(pool)) + .await + .expect("create server transport") +} + +/// Serve `handler` over `transport` on a spawned task. The returned receiver fires once `serve()` +/// has resolved `Ok`, i.e. the rmcp handshake completed and the service is running. +fn spawn_server( + handler: SurvivalServer, + transport: NostrServerTransport, +) -> ( + tokio::task::JoinHandle<()>, + tokio::sync::oneshot::Receiver<()>, +) { + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + let handle = tokio::spawn(async move { + match handler.serve(transport).await { + Ok(running) => { + let _ = ready_tx.send(()); + let _ = running.waiting().await; + } + Err(error) => { + eprintln!("server serve() failed: {error}"); + } + } + }); + (handle, ready_rx) +} + +/// A client transport talking plaintext to `server`, over its own pool in the shared network. +async fn client_transport(pool: &Arc, server: PublicKey) -> NostrClientTransport { + NostrClientTransport::with_relay_pool( + NostrClientTransportConfig::default() + .with_server_pubkey(server.to_hex()) + .with_encryption_mode(EncryptionMode::Disabled) + .with_relay_urls(vec!["wss://mock.relay".to_string()]), + as_pool(pool), + ) + .await + .expect("create client transport") +} + +/// The core survival assertion: a fresh, well-behaved client completes the MCP handshake and gets +/// a `tools/list` answered. On a dead server both time out. +async fn assert_client_can_serve_and_list(pool: &Arc, server: PublicKey) { + let transport = client_transport(pool, server).await; + let client = tokio::time::timeout( + Duration::from_secs(5), + SurvivalClient::latest().serve(transport), + ) + .await + .expect("client serve() timed out: the server transport is dead") + .expect("client serve() failed"); + + tokio::time::timeout(Duration::from_secs(5), client.list_all_tools()) + .await + .expect("list_all_tools timed out: the server transport is dead") + .expect("list_all_tools failed"); + + client.cancel().await.expect("client cancel"); +} + +/// Publish `hostile` from a stranger's key as the first message the server ever sees, then +/// require a fresh, well-behaved client to still connect and get a `tools/list` answered. +async fn assert_survives_hostile_first_message(hostile: JsonRpcMessage) { + let mut pools = MockRelayPool::create_linked_group(3); + let server_pool = Arc::new(pools.remove(0)); + let client_pool = Arc::new(pools.remove(0)); + let stranger_pool = Arc::new(pools.remove(0)); + let server_pubkey = server_pool.mock_public_key(); + + let transport = server_transport(&server_pool).await; + let (handle, _ready) = spawn_server(SurvivalServer, transport); + let_subscriptions_settle().await; + + publish_from(&stranger_pool, &Keys::generate(), server_pubkey, &hostile).await; + // Let the hostile event reach the handshake before the honest client's initialize does. + tokio::time::sleep(Duration::from_millis(50)).await; + + assert_client_can_serve_and_list(&client_pool, server_pubkey).await; + + handle.abort(); +} + +/// Poll the shared store until an event authored by `author` contains `needle`. +async fn wait_for_event_containing( + pool: &Arc, + author: PublicKey, + needle: &str, + within: Duration, +) -> Event { + let deadline = tokio::time::Instant::now() + within; + loop { + if let Some(event) = pool + .stored_events() + .await + .into_iter() + .find(|e| e.pubkey == author && e.content.contains(needle)) + { + return event; + } + if tokio::time::Instant::now() >= deadline { + panic!("no event from {author} containing {needle} within {within:?}"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// Poll the shared store until an event of `kind` exists. +async fn wait_for_kind(pool: &Arc, kind: u16, within: Duration) -> Event { + let deadline = tokio::time::Instant::now() + within; + loop { + if let Some(event) = pool + .stored_events() + .await + .into_iter() + .find(|e| e.kind == Kind::Custom(kind)) + { + return event; + } + if tokio::time::Instant::now() >= deadline { + panic!("no event of kind {kind} within {within:?}"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// The synthetic startup handshake is worker-local: nothing carrying its sentinel id may be +/// published. +async fn assert_sentinel_never_hit_the_wire(pool: &Arc) { + let leaked: Vec = pool + .stored_events() + .await + .into_iter() + .filter(|e| e.content.contains(STATELESS_SENTINEL)) + .map(|e| e.content) + .collect(); + assert!( + leaked.is_empty(), + "synthetic handshake messages must never reach the wire, found: {leaked:?}" + ); +} + +// ── Tests ───────────────────────────────────────────────────────────────── + +/// The worker satisfies the rmcp handshake itself, so `serve()` resolves with no client traffic at +/// all. Nothing else can satisfy it: without the worker's startup send, `serve()` stays pending +/// until some client happens to send an `initialize`. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn startup_service_ready_without_client_traffic() { + let server_pool = Arc::new(MockRelayPool::new()); + let transport = server_transport(&server_pool).await; + + let (handle, ready) = spawn_server(SurvivalServer, transport); + + tokio::time::timeout(Duration::from_secs(5), ready) + .await + .expect("serve() must resolve without any client traffic") + .expect("the server task ended before serve() resolved"); + + assert_sentinel_never_hit_the_wire(&server_pool).await; + + handle.abort(); +} + +/// One `notifications/initialized` from an unauthenticated stranger, before any request. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stranger_notification_before_init_survives() { + assert_survives_hostile_first_message(initialized_notification()).await; +} + +/// A response shape (`{"id":123,"result":{}}`) as the first inbound message. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn response_shape_before_init_survives() { + assert_survives_hostile_first_message(JsonRpcMessage::Response(JsonRpcResponse { + jsonrpc: "2.0".to_string(), + id: serde_json::json!(123), + result: serde_json::json!({}), + })) + .await; +} + +/// An error-response shape as the first inbound message (the same handshake arm as a response). +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn error_response_before_init_survives() { + assert_survives_hostile_first_message(JsonRpcMessage::ErrorResponse(JsonRpcErrorResponse { + jsonrpc: "2.0".to_string(), + id: serde_json::json!(1), + error: JsonRpcError { + code: -32000, + message: "x".to_string(), + data: None, + }, + })) + .await; +} + +/// No attacker: one honest client whose `notifications/initialized` overtakes its own +/// `initialize`, which Nostr permits since it does not order delivery. Driven on the raw wire so +/// the test picks the order. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn reordered_initialized_before_initialize_survives() { + let mut pools = MockRelayPool::create_linked_group(2); + let server_pool = Arc::new(pools.remove(0)); + let client_pool = Arc::new(pools.remove(0)); + let server_pubkey = server_pool.mock_public_key(); + + let transport = server_transport(&server_pool).await; + let (handle, _ready) = spawn_server(SurvivalServer, transport); + let_subscriptions_settle().await; + + let honest = Keys::generate(); + publish_from( + &client_pool, + &honest, + server_pubkey, + &initialized_notification(), + ) + .await; + publish_from( + &client_pool, + &honest, + server_pubkey, + &initialize_request("reorder-init"), + ) + .await; + publish_from( + &client_pool, + &honest, + server_pubkey, + &request("reorder-tools", "tools/list"), + ) + .await; + + // The client's own request id is restored on the response, so it is the needle. + wait_for_event_containing( + &server_pool, + server_pubkey, + "reorder-tools", + Duration::from_secs(5), + ) + .await; + + handle.abort(); +} + +/// The same stranger notification, but with one inbound middleware registered, which routes every +/// message through a detached per-message task instead of the event loop's inline path. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stranger_notification_with_middleware_survives() { + let mut pools = MockRelayPool::create_linked_group(3); + let server_pool = Arc::new(pools.remove(0)); + let client_pool = Arc::new(pools.remove(0)); + let stranger_pool = Arc::new(pools.remove(0)); + let server_pubkey = server_pool.mock_public_key(); + + let mut transport = server_transport(&server_pool).await; + transport.add_inbound_middleware(Arc::new(ForwardAll)); + let (handle, _ready) = spawn_server(SurvivalServer, transport); + let_subscriptions_settle().await; + + let stranger = Keys::generate(); + publish_from( + &stranger_pool, + &stranger, + server_pubkey, + &initialized_notification(), + ) + .await; + tokio::time::sleep(Duration::from_millis(50)).await; + + assert_client_can_serve_and_list(&client_pool, server_pubkey).await; + + handle.abort(); +} + +/// The first client's `initialize` no longer runs rmcp's pre-service handshake, so it is answered +/// with the handler's declared protocol version rather than an echo of the version it requested. +/// Assert on the FIRST client. Every later client is answered by the already-running service, +/// which never negotiates, so a later client reads the declared version no matter what the +/// handshake does and would pin nothing. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn first_client_older_protocol_version_not_negotiated() { + let mut pools = MockRelayPool::create_linked_group(2); + let server_pool = Arc::new(pools.remove(0)); + let client_pool = Arc::new(pools.remove(0)); + let server_pubkey = server_pool.mock_public_key(); + + let transport = server_transport(&server_pool).await; + let (handle, _ready) = spawn_server(SurvivalServer, transport); + let_subscriptions_settle().await; + + let transport = client_transport(&client_pool, server_pubkey).await; + let client = tokio::time::timeout( + Duration::from_secs(5), + SurvivalClient::requesting(ProtocolVersion::V_2024_11_05).serve(transport), + ) + .await + .expect("first client serve() timed out") + .expect("first client serve() failed"); + + let peer = client.peer_info().expect("peer info after serve()"); + assert_eq!( + peer.protocol_version, + ProtocolVersion::V_2025_11_25, + "the first client must receive the handler's declared protocol version, not an echo of \ + the version it requested" + ); + + client.cancel().await.expect("client cancel"); + handle.abort(); +} + +/// The announcement carries the handler's `protocolVersion` and the capability lists follow it. +/// The worker's startup handshake must not disturb either, which is what this pins. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn announced_server_publishes_capabilities() { + let server_pool = Arc::new(MockRelayPool::new()); + let transport = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default().with_announced_server(true), + as_pool(&server_pool), + ) + .await + .expect("create announced server transport"); + + let (handle, _ready) = spawn_server(SurvivalServer, transport); + + let announcement = wait_for_kind( + &server_pool, + SERVER_ANNOUNCEMENT_KIND, + Duration::from_secs(5), + ) + .await + .content; + let parsed: serde_json::Value = + serde_json::from_str(&announcement).expect("announcement content is JSON"); + assert_eq!( + parsed.get("protocolVersion").and_then(|v| v.as_str()), + Some(mcp_protocol_version()), + "announcement protocolVersion changed: {announcement}" + ); + + // The capability lists still follow the announcement. + wait_for_kind(&server_pool, TOOLS_LIST_KIND, Duration::from_secs(5)).await; + + assert_sentinel_never_hit_the_wire(&server_pool).await; + + handle.abort(); +}