fix: address review feedback — remove dead Conflict variant, restore sync on resume, rename test

- Remove unused SessionError::Conflict and dead string-matching in tower.rs
  (leftover from abandoned 409 approach)
- Restore sync() replay when replacing a dead primary common channel so
  server-initiated requests and cached notifications are not lost on reconnect
- Rename test from test_sse_channel_replacement_bug to test_sse_concurrent_streams
  per reviewer suggestion (describe what tests verify, not what triggered them)
- Add test for cache replay on dead primary replacement
- Use generic "MCP clients" in comments instead of specific client names

Signed-off-by: Mohammod Al Amin Ashik <maa.ashik00@gmail.com>

Author: its-mash

crates/rmcp/Cargo.toml

@@ -243,6 +243,6 @@ required-features = [
 path = "tests/test_custom_headers.rs"
 
 [[test]]
-name = "test_sse_channel_replacement_bug"
+name = "test_sse_concurrent_streams"
 required-features = ["server", "client", "transport-streamable-http-server", "transport-streamable-http-client", "reqwest"]
-path = "tests/test_sse_channel_replacement_bug.rs"
+path = "tests/test_sse_concurrent_streams.rs"

crates/rmcp/src/transport/streamable_http_server/session/local.rs

@@ -295,9 +295,9 @@ pub struct LocalSessionWorker {
     common: CachedTx,
     /// Shadow senders for secondary SSE streams (e.g. from POST EventSource
     /// reconnections). These keep the HTTP connections alive via SSE keep-alive
-    /// without receiving notifications, preventing clients like Cursor from
-    /// entering infinite reconnect loops when multiple EventSource connections
-    /// compete to replace the common channel.
+    /// without receiving notifications, preventing MCP clients from entering
+    /// infinite reconnect loops when multiple EventSource connections compete
+    /// to replace the common channel.
     shadow_txs: Vec<Sender<ServerSseMessage>>,
     event_rx: Receiver<SessionEvent>,
     session_config: SessionConfig,
@@ -321,8 +321,6 @@ pub enum SessionError {
     SessionServiceTerminated,
     #[error("Invalid event id")]
     InvalidEventId,
-    #[error("Conflict: Only one standalone SSE stream is allowed per session")]
-    Conflict,
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
 }
@@ -546,29 +544,37 @@ impl LocalSessionWorker {
                         http_request_id,
                         "Request-wise channel completed, falling back to common channel"
                     );
-                    self.resume_or_shadow_common()
+                    self.resume_or_shadow_common(last_event_id.index).await
                 }
             }
-            None => self.resume_or_shadow_common(),
+            None => self.resume_or_shadow_common(last_event_id.index).await,
         }
     }
 
     /// Resume the common channel, or create a shadow stream if the primary is
     /// still active.
     ///
     /// When the primary common channel is dead (receiver dropped), replace it
-    /// so this stream becomes the new primary notification channel.
+    /// so this stream becomes the new primary notification channel. Cached
+    /// messages are replayed from `last_event_index` so the client receives
+    /// any events it missed (including server-initiated requests).
     ///
     /// When the primary is still active, create a "shadow" stream — an idle SSE
     /// connection kept alive by keep-alive pings. This prevents multiple
     /// EventSource connections (e.g. from POST response reconnections) from
     /// killing each other by repeatedly replacing the common channel sender.
-    fn resume_or_shadow_common(&mut self) -> Result<StreamableHttpMessageReceiver, SessionError> {
+    async fn resume_or_shadow_common(
+        &mut self,
+        last_event_index: usize,
+    ) -> Result<StreamableHttpMessageReceiver, SessionError> {
         let (tx, rx) = tokio::sync::mpsc::channel(self.session_config.channel_capacity);
         if self.common.tx.is_closed() {
             // Primary common channel is dead — replace it.
             tracing::debug!("Replacing dead common channel with new primary");
             self.common.tx = tx;
+            // Replay cached messages from where the client left off so
+            // server-initiated requests and notifications are not lost.
+            self.common.sync(last_event_index).await?;
         } else {
             // Primary common channel is still active. Create a shadow stream
             // that stays alive via SSE keep-alive but doesn't receive

crates/rmcp/src/transport/streamable_http_server/tower.rs

@@ -328,20 +328,11 @@ where
             .map(|s| s.to_owned());
         if let Some(last_event_id) = last_event_id {
             // check if session has this event id
-            let stream = match self
+            let stream = self
                 .session_manager
                 .resume(&session_id, last_event_id)
                 .await
-            {
-                Ok(stream) => stream,
-                Err(e) if e.to_string().contains("Conflict:") => {
-                    return Ok(Response::builder()
-                        .status(http::StatusCode::CONFLICT)
-                        .body(Full::new(Bytes::from(e.to_string())).boxed())
-                        .expect("valid response"));
-                }
-                Err(e) => return Err(internal_error_response("resume session")(e)),
-            };
+                .map_err(internal_error_response("resume session"))?;
             // Resume doesn't need priming - client already has the event ID
             Ok(sse_stream_response(
                 stream,
@@ -350,20 +341,11 @@ where
             ))
         } else {
             // create standalone stream
-            let stream = match self
+            let stream = self
                 .session_manager
                 .create_standalone_stream(&session_id)
                 .await
-            {
-                Ok(stream) => stream,
-                Err(e) if e.to_string().contains("Conflict:") => {
-                    return Ok(Response::builder()
-                        .status(http::StatusCode::CONFLICT)
-                        .body(Full::new(Bytes::from(e.to_string())).boxed())
-                        .expect("valid response"));
-                }
-                Err(e) => return Err(internal_error_response("create standalone stream")(e)),
-            };
+                .map_err(internal_error_response("create standalone stream"))?;
             // Prepend priming event if sse_retry configured
             let stream = if let Some(retry) = self.config.sse_retry {
                 let priming = ServerSseMessage {

…ests/test_sse_channel_replacement_bug.rs …mcp/tests/test_sse_concurrent_streams.rscrates/rmcp/tests/test_sse_channel_replacement_bug.rs renamed to crates/rmcp/tests/test_sse_concurrent_streams.rs crates/rmcp/tests/test_sse_channel_replacement_bug.rs renamed to crates/rmcp/tests/test_sse_concurrent_streams.rs

@@ -1,12 +1,12 @@
-/// Tests for SSE channel replacement fix (shadow channels)
+/// Tests for concurrent SSE stream handling (shadow channels)
 ///
 /// These tests verify that multiple GET SSE streams on the same session
 /// don't kill each other by replacing the common channel sender.
 ///
-/// Root cause: When POST SSE responses include `retry`, EventSource reconnects
-/// via GET after the stream ends. Each GET was unconditionally replacing
-/// `self.common.tx`, killing the other stream's receiver — causing an infinite
-/// reconnect loop every `sse_retry` seconds.
+/// Root cause: When POST SSE responses include `retry`, the EventSource API
+/// reconnects via GET after the stream ends. Each GET was unconditionally
+/// replacing `self.common.tx`, killing the other stream's receiver — causing
+/// an infinite reconnect loop every `sse_retry` seconds.
 ///
 /// Fix: `resume_or_shadow_common()` checks if the primary common channel is
 /// still active. If so, it creates a "shadow" stream (idle, keep-alive only)
@@ -700,3 +700,46 @@ async fn dropping_shadows_does_not_affect_primary() {
 
     ct.cancel();
 }
+
+// ─── Tests: Cache replay on dead primary replacement ─────────────────────────
+
+/// When a notification is sent while the primary is alive, then the primary
+/// dies and a new GET resumes with Last-Event-ID "0", the replacement primary
+/// should receive the cached notification via sync() replay.
+#[tokio::test]
+async fn dead_primary_replacement_replays_cached_events() {
+    let ct = CancellationToken::new();
+    let trigger = Arc::new(Notify::new());
+    let url = start_test_server(ct.clone(), trigger.clone()).await;
+    let client = reqwest::Client::new();
+
+    let session_id = initialize_session(&client, &url).await;
+    send_initialized_notification(&client, &url, &session_id).await;
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // First GET — becomes primary
+    let get1 = open_standalone_get(&client, &url, &session_id).await;
+    assert_eq!(get1.status(), 200);
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // Trigger notification while primary is alive (gets cached)
+    trigger.notify_one();
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Drop primary — notification was sent and cached
+    drop(get1);
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // Resume with Last-Event-ID "0" — primary is dead, should replace it
+    // and replay cached events from index 0
+    let get_resume = open_resume_get(&client, &url, &session_id, "0").await;
+    assert_eq!(get_resume.status(), 200);
+
+    // The cached notification should be replayed on the new primary
+    assert!(
+        wait_for_sse_event(get_resume, "tools/list_changed", Duration::from_secs(3)).await,
+        "Replacement primary should receive cached notification via sync() replay"
+    );
+
+    ct.cancel();
+}