Merge pull request #532 from koic/activate_listen_streams_after_the_acknowledgement

Deliver `subscriptions/listen` notifications only after the acknowledgement

Author: koic

lib/mcp/server/transports/streamable_http_transport.rb

@@ -169,7 +169,7 @@ def initialize(
           @allowed_origins = Array(allowed_origins).map(&:downcase).freeze
           @pending_responses = {}
 
-          # Maps a `subscriptions/listen` request id to `{ stream: stream_object, filter: honored_subscription_filter }` (SEP-2575).
+          # Maps a `subscriptions/listen` request id to `{ stream: stream_object, filter: honored_subscription_filter, active: boolean }` (SEP-2575).
           # In-process only; a multi-worker deployment needs an external event bus to fan notifications out across processes,
           # which is a follow-up.
           @listen_subscriptions = {}
@@ -922,6 +922,12 @@ def first
 
         # The body registers the stream and returns, leaving the response open like
         # the legacy GET stream (`create_sse_body`).
+        #
+        # Registration and activation are split on purpose: the entry is inserted inactive
+        # (reserving the id and the cap slot atomically), the acknowledgement is written outside the lock,
+        # and only then does the entry become eligible for delivery. A concurrent notification between
+        # the insert and the acknowledgement write skips the inactive entry,
+        # enforcing the SEP-2575 rule that no notification precedes the acknowledgement.
         def listen_sse_body(request_id, honored)
           ListenStreamBody.new do |stream|
             rejected = false
@@ -930,7 +936,7 @@ def listen_sse_body(request_id, honored)
                   (@max_listen_subscriptions && @listen_subscriptions.size >= @max_listen_subscriptions)
                 rejected = true
               else
-                @listen_subscriptions[request_id] = { stream: stream, filter: honored }
+                @listen_subscriptions[request_id] = { stream: stream, filter: honored, active: false }
               end
             end
 
@@ -948,6 +954,7 @@ def listen_sse_body(request_id, honored)
 
               begin
                 send_to_stream(stream, acknowledgement)
+                activate_listen_subscription(request_id)
                 start_listen_keepalive_thread(request_id)
               rescue *STREAM_WRITE_ERRORS
                 remove_listen_subscription(request_id)
@@ -957,6 +964,15 @@ def listen_sse_body(request_id, honored)
           end
         end
 
+        # Marks a listen subscription eligible for delivery once its acknowledgement write has completed.
+        # The entry may already be gone when the transport closed concurrently.
+        def activate_listen_subscription(request_id)
+          @mutex.synchronize do
+            subscription = @listen_subscriptions[request_id]
+            subscription[:active] = true if subscription
+          end
+        end
+
         # Periodically writes an SSE keepalive comment frame to a listen stream so a silently dropped
         # connection is detected and its slot freed, rather than held until the next fan-out write.
         # Mirrors the legacy GET stream's `start_keepalive_thread`; a comment frame (not a data frame)
@@ -1038,6 +1054,10 @@ def deliver_to_listen_subscriptions(method, params)
           # a slow or stalled subscriber must not block the transport, matching the legacy delivery paths.
           matched = @mutex.synchronize do
             @listen_subscriptions.filter_map do |request_id, subscription|
+              # An inactive entry has not finished writing its acknowledgement yet;
+              # delivering to it would put a notification ahead of the acknowledgement.
+              next unless subscription[:active]
+
               hit = if field
                 subscription[:filter][field]
               else

test/mcp/server/transports/streamable_http_transport_test.rb

@@ -6108,6 +6108,24 @@ def string
           transport.close
         end
 
+        test "notifications are delivered only after the listen acknowledgement" do
+          # Inject an entry in the registered-but-not-yet-acknowledged state: the window between
+          # the registry insert and the acknowledgement write, which happens outside the lock.
+          io = StringIO.new
+          @transport.instance_variable_get(:@listen_subscriptions)["listen-1"] = {
+            stream: io, filter: { toolsListChanged: true }, active: false
+          }
+
+          @server.notify_tools_list_changed
+
+          assert_empty sse_events(io)
+
+          @transport.send(:activate_listen_subscription, "listen-1")
+          @server.notify_tools_list_changed
+
+          assert_equal ["notifications/tools/list_changed"], sse_events(io).map { |event| event["method"] }
+        end
+
         test "subscriptions/listen streams for different subscriptions receive their own subscriptionId" do
           first = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true })
           second = open_listen_stream(id: "listen-2", notifications: { toolsListChanged: true })