test: harden SEP-2260 stream-enforcement e2e (#1033)

Detect handler invocation via a channel asserted empty instead of a
panic in a spawned task (swallowed, cannot fail the test); surface
scripted-server misuse as transport errors rather than panics in the
transport task; bound the tail awaits with 5s timeouts. Correct the
header comment: a 2026-07-28 server minting a session id is not
spec-legal (SEP-2567 removes sessions and the GET endpoint) — the
scripted server is deliberately non-conforming, which is the point of
receive-side enforcement.

Author: gocamille

crates/rmcp/tests/test_sep_2260_stream_enforcement.rs

@@ -1,16 +1,23 @@
 //! SEP-2260 follow-up (#1033): stream-based receive-side enforcement.
 //!
 //! Scripted streamable HTTP "server": answers a legacy initialize with
-//! protocol 2026-07-28 and a session id (spec-legal; rmcp's own server is
-//! stateless at that version, but the client must be correct against any
-//! server), so the client has BOTH a standalone GET stream and strict
+//! protocol 2026-07-28 AND a session id. That combination is NOT
+//! spec-compliant: the 2026-07-28 revision removes protocol-level sessions
+//! and the standalone GET stream (SEP-2567; transports spec: "do not mint
+//! or echo session IDs"). Receive-side enforcement (#1033) exists precisely
+//! to protect the client from non-conforming servers, and rmcp's client
+//! tolerates the session id and opens the standalone GET stream — so this
+//! is the reachable path where the client has BOTH a GET stream and strict
 //! SEP-2260 enforcement.
 #![cfg(all(
     feature = "client",
     feature = "transport-streamable-http-client",
     not(feature = "local")
 ))]
-#![allow(deprecated)]
+#![expect(
+    deprecated,
+    reason = "Sampling is deprecated by SEP-2577 but remains the canonical restricted request"
+)]
 
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
@@ -70,7 +77,8 @@ impl StreamableHttpClient for ScriptedServer {
         _custom_headers: HashMap<HeaderName, HeaderValue>,
     ) -> Result<StreamableHttpPostResponse, StreamableHttpError<Self::Error>> {
         let value = serde_json::to_value(&message).expect("serialize client message");
-        self.posted.send(value.clone()).expect("test alive");
+        // Receiver drop is normal at test teardown; never panic in the transport task.
+        let _ = self.posted.send(value.clone());
         if value["method"] == "initialize" {
             let mut info = ServerInfo::new(ServerCapabilities::default());
             info.protocol_version = ProtocolVersion::V_2026_07_28;
@@ -84,12 +92,14 @@ impl StreamableHttpClient for ScriptedServer {
             ));
         }
         if matches!(message, ClientJsonRpcMessage::Request(_)) {
-            let rx = self
-                .post_stream
-                .lock()
-                .await
-                .take()
-                .expect("exactly one non-initialize request POST in this test");
+            // Fail as a transport error rather than panicking: this code runs
+            // in the transport task, where a panic is swallowed and shows up
+            // only as an opaque timeout in the test.
+            let rx = self.post_stream.lock().await.take().ok_or_else(|| {
+                StreamableHttpError::Client(std::io::Error::other(
+                    "scripted server expects exactly one non-initialize request POST",
+                ))
+            })?;
             return Ok(StreamableHttpPostResponse::Sse(message_stream(rx), None));
         }
         Ok(StreamableHttpPostResponse::Accepted)
@@ -123,8 +133,10 @@ impl StreamableHttpClient for ScriptedServer {
 
 #[derive(Clone)]
 struct SamplingClient {
-    // Some(tx): forward sampling params; None: sampling must never reach the handler.
-    on_sampling: Option<mpsc::UnboundedSender<CreateMessageRequestParams>>,
+    // Every invocation is recorded here. Tests assert on the receiver (the
+    // negative test asserts it stays empty); a panic in this handler would
+    // run in a spawned task and be silently swallowed.
+    sampled: mpsc::UnboundedSender<CreateMessageRequestParams>,
 }
 
 impl ClientHandler for SamplingClient {
@@ -133,10 +145,7 @@ impl ClientHandler for SamplingClient {
         params: CreateMessageRequestParams,
         _context: RequestContext<RoleClient>,
     ) -> Result<CreateMessageResult, rmcp::ErrorData> {
-        let Some(tx) = &self.on_sampling else {
-            panic!("sampling request must not reach the handler in this test");
-        };
-        tx.send(params).expect("test alive");
+        let _ = self.sampled.send(params);
         Ok(CreateMessageResult::new(
             SamplingMessage::assistant_text("pong"),
             "test-model".to_string(),
@@ -171,21 +180,27 @@ async fn next_posted(posted: &mut mpsc::UnboundedReceiver<Value>) -> Value {
         .expect("channel open")
 }
 
-/// Drive startup + one in-flight tools/list; return (client, posted rx,
-/// get stream tx, post stream tx, in-flight call handle, tools/list id).
-async fn setup(
-    handler: SamplingClient,
-) -> (
-    rmcp::service::RunningService<RoleClient, SamplingClient>,
-    mpsc::UnboundedReceiver<Value>,
-    mpsc::Sender<Value>,
-    mpsc::Sender<Value>,
-    tokio::task::JoinHandle<Result<rmcp::model::ListToolsResult, rmcp::ServiceError>>,
-    Value,
-) {
+struct Harness {
+    client: rmcp::service::RunningService<RoleClient, SamplingClient>,
+    /// Every message the client POSTs to the scripted server.
+    posted: mpsc::UnboundedReceiver<Value>,
+    /// Feeds the standalone GET stream.
+    get_tx: mpsc::Sender<Value>,
+    /// Feeds the SSE stream of the in-flight tools/list POST.
+    post_tx: mpsc::Sender<Value>,
+    /// In-flight tools/list call (response withheld until the test releases it).
+    call: tokio::task::JoinHandle<Result<rmcp::model::ListToolsResult, rmcp::ServiceError>>,
+    tools_list_id: Value,
+    /// Sampling params seen by the client handler.
+    sampled: mpsc::UnboundedReceiver<CreateMessageRequestParams>,
+}
+
+/// Drive startup + one in-flight tools/list.
+async fn setup() -> Harness {
     let (get_tx, get_rx) = mpsc::channel(8);
     let (post_tx, post_rx) = mpsc::channel(8);
     let (posted_tx, mut posted_rx) = mpsc::unbounded_channel();
+    let (sampled_tx, sampled_rx) = mpsc::unbounded_channel();
     let server = ScriptedServer {
         get_stream: Arc::new(Mutex::new(Some(get_rx))),
         post_stream: Arc::new(Mutex::new(Some(post_rx))),
@@ -195,9 +210,15 @@ async fn setup(
         server,
         StreamableHttpClientTransportConfig::with_uri("http://scripted/mcp"),
     );
-    let client = serve_client_with_lifecycle(handler, transport, ClientLifecycleMode::Initialize)
-        .await
-        .expect("initialize against scripted server");
+    let client = serve_client_with_lifecycle(
+        SamplingClient {
+            sampled: sampled_tx,
+        },
+        transport,
+        ClientLifecycleMode::Initialize,
+    )
+    .await
+    .expect("initialize against scripted server");
 
     // initialize + notifications/initialized already posted during startup.
     assert_eq!(next_posted(&mut posted_rx).await["method"], "initialize");
@@ -213,7 +234,15 @@ async fn setup(
     assert_eq!(tools_list["method"], "tools/list");
     let tools_list_id = tools_list["id"].clone();
 
-    (client, posted_rx, get_tx, post_tx, call, tools_list_id)
+    Harness {
+        client,
+        posted: posted_rx,
+        get_tx,
+        post_tx,
+        call,
+        tools_list_id,
+        sampled: sampled_rx,
+    }
 }
 
 /// #1033 scenario 1: a restricted request on the standalone GET stream while
@@ -222,12 +251,11 @@ async fn setup(
 #[tokio::test]
 async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_flight()
 -> anyhow::Result<()> {
-    let (client, mut posted_rx, get_tx, post_tx, call, tools_list_id) =
-        setup(SamplingClient { on_sampling: None }).await;
+    let mut h = setup().await;
 
-    get_tx.send(sampling_request(100)).await?;
+    h.get_tx.send(sampling_request(100)).await?;
 
-    let rejection = next_posted(&mut posted_rx).await;
+    let rejection = next_posted(&mut h.posted).await;
     assert_eq!(
         rejection["id"], 100,
         "reply to the sampling request: {rejection}"
@@ -238,25 +266,28 @@ async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_fl
          request in flight, got {rejection}"
     );
 
-    post_tx.send(tools_list_response(&tools_list_id)).await?;
-    call.await??;
-    client.cancel().await?;
+    h.post_tx
+        .send(tools_list_response(&h.tools_list_id))
+        .await?;
+    tokio::time::timeout(Duration::from_secs(5), h.call).await???;
+    // Rejected means rejected: the handler must not ALSO have been invoked.
+    assert!(
+        h.sampled.try_recv().is_err(),
+        "handler must not see the rejected sampling request"
+    );
+    tokio::time::timeout(Duration::from_secs(5), h.client.cancel()).await??;
     Ok(())
 }
 
 /// Positive twin: the same restricted request arriving on the SSE stream of
 /// the originating POST is dispatched to the handler and answered.
 #[tokio::test]
 async fn restricted_request_on_originating_post_stream_is_dispatched() -> anyhow::Result<()> {
-    let (sampled_tx, mut sampled_rx) = mpsc::unbounded_channel();
-    let (client, mut posted_rx, _get_tx, post_tx, call, tools_list_id) = setup(SamplingClient {
-        on_sampling: Some(sampled_tx),
-    })
-    .await;
+    let mut h = setup().await;
 
-    post_tx.send(sampling_request(200)).await?;
+    h.post_tx.send(sampling_request(200)).await?;
 
-    let response = next_posted(&mut posted_rx).await;
+    let response = next_posted(&mut h.posted).await;
     assert_eq!(
         response["id"], 200,
         "reply to the sampling request: {response}"
@@ -265,12 +296,14 @@ async fn restricted_request_on_originating_post_stream_is_dispatched() -> anyhow
         response["result"]["model"], "test-model",
         "request on the originating POST stream must reach the handler, got {response}"
     );
-    tokio::time::timeout(Duration::from_secs(5), sampled_rx.recv())
+    tokio::time::timeout(Duration::from_secs(5), h.sampled.recv())
         .await?
         .expect("handler invoked");
 
-    post_tx.send(tools_list_response(&tools_list_id)).await?;
-    call.await??;
-    client.cancel().await?;
+    h.post_tx
+        .send(tools_list_response(&h.tools_list_id))
+        .await?;
+    tokio::time::timeout(Duration::from_secs(5), h.call).await???;
+    tokio::time::timeout(Duration::from_secs(5), h.client.cancel()).await??;
     Ok(())
 }