test(sep-2567): pin the replacement handshake and its teardown

Records the shutdown DELETE instead of parsing its empty body as JSON, so the
teardown is assertable rather than a panic in the server task.

Author: LizunovSergey

crates/rmcp/tests/test_streamable_http_sessionless_version.rs

@@ -98,10 +98,21 @@ impl Recorder {
             .collect()
     }
 
+    fn delete_requests(&self) -> Vec<Call> {
+        self.calls()
+            .into_iter()
+            .filter(|(http_method, ..)| http_method == "DELETE")
+            .collect()
+    }
+
+    /// Session headers on everything after the handshake that is not the teardown DELETE,
+    /// which carries the id on purpose so the server tears the session down.
     fn session_headers_after_handshake(&self) -> Vec<Option<String>> {
         self.calls()
             .into_iter()
-            .filter(|(_, jsonrpc_method, _)| jsonrpc_method != "initialize")
+            .filter(|(http_method, jsonrpc_method, _)| {
+                jsonrpc_method != "initialize" && http_method != "DELETE"
+            })
             .map(|(.., session)| session)
             .collect()
     }
@@ -144,6 +155,19 @@ async fn handler(
             .expect("build GET rejection");
     }
 
+    // The shutdown DELETE carries no body, so record it before anything parses one.
+    if method == axum::http::Method::DELETE {
+        state.seen.lock().expect("recorder poisoned").push((
+            "DELETE".to_owned(),
+            "-".to_owned(),
+            session,
+        ));
+        return Response::builder()
+            .status(StatusCode::OK)
+            .body(Body::empty())
+            .expect("build session teardown response");
+    }
+
     let request: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON-RPC body");
     let jsonrpc_method = request["method"].as_str().unwrap_or("-").to_owned();
     state.seen.lock().expect("recorder poisoned").push((
@@ -347,7 +371,7 @@ async fn replacement_handshake_at_a_modern_version_drops_the_session_too() {
     // Nor on anything after it, and no stream on the replacement session either.
     let echoed: Vec<_> = replacement
         .iter()
-        .filter(|(.., session)| session.is_some())
+        .filter(|(http_method, _, session)| session.is_some() && http_method != "DELETE")
         .collect();
     assert!(
         echoed.is_empty(),
@@ -359,6 +383,16 @@ async fn replacement_handshake_at_a_modern_version_drops_the_session_too() {
         "expected only the legacy session's GET stream, got {:?}",
         recorder.get_requests()
     );
+
+    // Dropping the id is a request-and-stream decision, not a licence to leak server
+    // state: the session the server really did create is still torn down at shutdown.
+    let deletes = recorder.delete_requests();
+    assert!(
+        deletes
+            .iter()
+            .any(|(.., session)| session.as_deref() == Some(REPLACEMENT_SESSION_ID)),
+        "the replacement session was never deleted at shutdown: {deletes:?}"
+    );
 }
 
 #[tokio::test]