diff --git a/conformance/expected_failures.yml b/conformance/expected_failures.yml index f21d5598..438c8cc5 100644 --- a/conformance/expected_failures.yml +++ b/conformance/expected_failures.yml @@ -1,8 +1,4 @@ server: - # SEP-1699 resumability: the server sends no priming event or `retry` field on a POST SSE stream, and - # the referee warns about each of the two; the baseline counts a warning as a failure. The resume on - # `Last-Event-ID` is not exercised, since the server does not close the stream mid-call. - - server-sse-polling # SEP-2663 (io.modelcontextprotocol/tasks): the server does not implement the tasks extension. # `tasks-status-notifications` is intentionally absent: the referee runs no check in it, and a baseline entry # for a scenario with no failing check is flagged stale. diff --git a/conformance/server.rb b/conformance/server.rb index 42b5f2b8..f849d4df 100644 --- a/conformance/server.rb +++ b/conformance/server.rb @@ -6,6 +6,7 @@ require "set" require "uri" require_relative "../lib/mcp" +require_relative "../examples/in_memory_event_store" module Conformance # 1x1 red PNG pixel (matches TypeScript SDK and Python SDK) @@ -319,7 +320,13 @@ class TestReconnection < MCP::Tool description "A tool that triggers SSE stream closure to test client reconnection behavior" class << self - def call(**_args) + # SEP-1699: the response stream is closed mid-call, so the referee has to reconnect with `Last-Event-ID` + # to receive the second log message and the result, as the TypeScript and Python fixtures do. + def call(server_context:, **_args) + server_context.notify_log_message(data: "Before disconnect", level: "info", logger: "test_reconnection") + server_context.close_sse_stream + sleep(0.2) + server_context.notify_log_message(data: "After reconnect", level: "info", logger: "test_reconnection") MCP::Tool::Response.new([MCP::Content::Text.new("Reconnection test completed").to_h]) end end @@ -909,7 +916,12 @@ def initialize(port: DEFAULT_PORT) def start server = build_server - transport = MCP::Server::Transports::StreamableHTTPTransport.new(server) + transport = MCP::Server::Transports::StreamableHTTPTransport.new( + server, + # SEP-1699: lets `test_reconnection` close its stream and the referee resume it with `Last-Event-ID`. + event_store: InMemoryEventStore.new, + retry_interval: 100, + ) configure_handlers(server) rack_app = build_rack_app(transport) diff --git a/conformance/server_runner.rb b/conformance/server_runner.rb index bdf61bcb..208cd5d6 100644 --- a/conformance/server_runner.rb +++ b/conformance/server_runner.rb @@ -12,6 +12,15 @@ class ServerRunner SERVER_POLL_INTERVAL = 0.5 SERVER_HEALTH_CHECK_RETRIES = (SERVER_START_TIMEOUT / SERVER_POLL_INTERVAL).to_i + # The referee runs on whatever Node `npx` finds. On Node 26, `fetch` rejects a promise without a reason when + # a response body is cancelled on a pooled connection while the server's next frame is still in flight, + # which the `server-sse-multiple-streams` scenario does right after reading a stream's priming event; nothing is + # reported by that rejection, but it ends the referee process mid-suite. Node 22 and 24 (the version CI runs) + # do not reject. The warning keeps the suite running; the referee's own checks decide the result as before. + # Appended after any `NODE_OPTIONS` the environment sets, so it wins over an `--unhandled-rejections` value given + # there: the referee is the only process this script starts, and a strict setting would end it the same way. + REFEREE_NODE_OPTIONS = "--unhandled-rejections=warn" + def initialize(port: Server::DEFAULT_PORT, scenario: nil, spec_version: nil, requirements: nil, verbose: false) @port = port @scenario = scenario @@ -87,7 +96,7 @@ def run_conformance(command, server_pid:) conformance_exit_code = nil begin - system(*command) + system(referee_env, *command) conformance_exit_code = $CHILD_STATUS.exitstatus ensure terminate_server(server_pid) @@ -96,6 +105,13 @@ def run_conformance(command, server_pid:) exit(conformance_exit_code || 1) unless conformance_exit_code == 0 end + # Appends the referee's Node options to any the environment already sets. + def referee_env + options = [ENV["NODE_OPTIONS"], REFEREE_NODE_OPTIONS].compact.reject(&:empty?) + + { "NODE_OPTIONS" => options.join(" ") } + end + def terminate_server(pid) Process.kill("TERM", pid) rescue Errno::ESRCH diff --git a/docs/_server/transports.md b/docs/_server/transports.md index 9bbcc985..6559d8b0 100644 --- a/docs/_server/transports.md +++ b/docs/_server/transports.md @@ -220,6 +220,72 @@ This mode is suitable for simple tool servers that do not need server-initiated > Requests of the [modern lifecycle](/server/discover/#the-stateless-modern-lifecycle) of MCP 2026-07-28 ignore `enable_json_response:` > and are served as SSE-framed single exchanges, so request-scoped notifications still reach the client. +### Resumable SSE Streams + +Per MCP 2025-11-25 (SEP-1699), a client whose SSE stream is disconnected can reconnect with a GET carrying +the `Last-Event-ID` header, and the server replays the events the client missed and resumes the stream from that point. +`StreamableHTTPTransport` does this once it is given an event store, which keeps each stream's events under the ids it assigns. +The SDK defines the contract of a store, `MCP::Server::Transports::EventStore`, and ships no store of its own; +[`examples/in_memory_event_store.rb`](https://github.com/modelcontextprotocol/ruby-sdk/blob/main/examples/in_memory_event_store.rb) +is one to copy into an application: + +```ruby +require_relative "in_memory_event_store" # copied from the SDK's examples + +transport = MCP::Server::Transports::StreamableHTTPTransport.new( + server, + event_store: InMemoryEventStore.new, + retry_interval: 1000, +) +``` + +With an event store, the SSE events the transport writes carry the `id:` the store assigned them, and each stream opens with a priming event: +an event with an id, the `retry` field when `retry_interval:` is set (the milliseconds the client waits before reconnecting), and empty data, +so the client has a cursor to resume from before anything else is sent. A client that negotiated a protocol version before 2025-11-25 +receives no priming event, since it cannot parse an event with empty data; its other events still carry ids. +An event the store failed to take is written without an id and the failure is reported, so the client's cursor stays at the event before it, +and the events of the session's GET stream are stored from the moment the client opens that stream, the earliest point it can hold a cursor into them. +A GET with `Last-Event-ID` resumes the stream the event belongs to, whether the session's GET stream or the response stream of a POST: +the events after that id are replayed, and when the request is still running, the new connection takes the old one's place, +so the handler's remaining notifications and its response reach the client. A GET stream whose connection fails is detached from the session +rather than ending it, so the client resumes it the same way. A resume whose replay passes keep finding events, because events keep being stored +while each batch is written, is closed after its sixteenth batch instead, and the client reconnects from its new cursor after `retry`. +The id must name an event of the requesting session; +an id of another session, or one the store no longer holds, is answered with HTTP 400 and nothing is replayed. + +The example `InMemoryEventStore` keeps events in the process, which suits the single-process deployments the session store needs anyway. +It is bounded: each stream keeps at most `max_events_per_stream:` events (default 1024), at most `max_streams:` streams are kept (default 1024), +and their messages take at most `max_bytes:` in all (default 64 MiB), apart from the stream written to last, which is never dropped; +the oldest events of a stream and the least recently written streams go first. +The store keeps what it holds, responses included, after the request completes and after the session ends, until the bounds drop it, +and the bounds are shared by every session of the transport, so the traffic of one session can evict the events another would resume from. +A deployment whose reconnects may reach another process implements `MCP::Server::Transports::EventStore` against a shared backend; +the class documents the two methods a store provides. + +A handler can also close its own connection on purpose, so a long-running operation does not hold one open: + +```ruby +class SlowTool < MCP::Tool + class << self + def call(server_context:) + server_context.close_sse_stream # the client reconnects after `retry_interval` + result = compute_for_a_while + MCP::Tool::Response.new([{ type: "text", text: result }]) + end + end +end +``` + +`close_sse_stream` returns `true` when it closed the connection of the current request's stream and `false` when there was nothing to close: +no event store, a client older than 2025-11-25, a request answered as JSON, or a connection whose priming event the store failed to take. +The request keeps running either way, +and its remaining messages are stored for the reconnect. `close_standalone_sse_stream` does the same for the session's GET stream. + +{: .note } +> Requests of the [modern lifecycle](/server/discover/#the-stateless-modern-lifecycle) (MCP 2026-07-28) are single exchanges +> whose streams are not resumable, as that revision specifies, so a `Last-Event-ID` header on them is ignored. +> Stateless mode (`stateless: true`) has no sessions to resume, so it does not accept `event_store:`. + ### Session Limits By default, stateful sessions are bounded in three ways: they expire after `session_idle_timeout` seconds of inactivity diff --git a/examples/README.md b/examples/README.md index 62dedd1c..938fa39f 100644 --- a/examples/README.md +++ b/examples/README.md @@ -207,6 +207,21 @@ $ ruby examples/modern_http_client.rb The same server still accepts the legacy `initialize` flow: the transport routes each request to the legacy or modern lifecycle by its `MCP-Protocol-Version` header. +### 10. In-Memory Event Store (`in_memory_event_store.rb`) + +An `MCP::Server::Transports::EventStore` for resumable SSE streams (MCP 2025-11-25, SEP-1699) that keeps events in the process. The SDK ships the contract of a store and no store of its own, +so this file is meant to be copied into an application. + +**Usage:** + +```ruby +require_relative "in_memory_event_store" + +transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, event_store: InMemoryEventStore.new) +``` + +It suits a single process and is bounded by events per stream, by streams, and by bytes. A deployment whose reconnects may reach another process needs a store on a shared backend instead. + ### Testing with MCP Inspector [MCP Inspector](https://modelcontextprotocol.io/docs/tools/inspector) is a browser-based tool for testing and debugging MCP servers. diff --git a/examples/in_memory_event_store.rb b/examples/in_memory_event_store.rb new file mode 100644 index 00000000..20f30d15 --- /dev/null +++ b/examples/in_memory_event_store.rb @@ -0,0 +1,102 @@ +# frozen_string_literal: true + +require "securerandom" +require "mcp" + +# An `MCP::Server::Transports::EventStore` that keeps events in the process, for the single-process deployments +# `MCP::Server::Transports::StreamableHTTPTransport` serves sessions from. The SDK ships the `EventStore` contract +# and no store of its own: copy this file into an application and pass an instance as `event_store:`. +# +# transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, event_store: InMemoryEventStore.new) +# +# The store is bounded in three ways, since it retains what handlers emit for clients that may never come back: +# each stream keeps at most `max_events_per_stream` events, dropping its oldest; at most `max_streams` streams are kept, +# and the total size of the retained messages stays under `max_bytes`, both by dropping the streams written to least recently. +# A dropped event can no longer be resumed from: a client naming it in `Last-Event-ID` is told so by the transport and has to start over. +# +# A deployment whose reconnects may reach another process needs a store on a shared backend instead. +class InMemoryEventStore < MCP::Server::Transports::EventStore + DEFAULT_MAX_EVENTS_PER_STREAM = 1024 + DEFAULT_MAX_STREAMS = 1024 + DEFAULT_MAX_BYTES = 64 * 1024 * 1024 + + def initialize(max_events_per_stream: DEFAULT_MAX_EVENTS_PER_STREAM, max_streams: DEFAULT_MAX_STREAMS, max_bytes: DEFAULT_MAX_BYTES) + super() + + { max_events_per_stream: max_events_per_stream, max_streams: max_streams, max_bytes: max_bytes }.each do |name, value| + raise ArgumentError, "#{name} must be a positive Integer" unless value.is_a?(Integer) && value.positive? + end + + @max_events_per_stream = max_events_per_stream + @max_streams = max_streams + @max_bytes = max_bytes + @mutex = Mutex.new + # Maps a stream id to `{ nonce: String, sequence: Integer, bytes: Integer, events: [[event_id, message], ...] }`, in order of + # the last write to each stream (a written stream is moved to the end), so the first entry is the one + # to drop when the bounds are reached. + @streams = {} + @total_bytes = 0 + end + + # Event ids are `/-`: the stream is found again from an id without an index, the sequence orders + # the events of a stream however many were dropped before the one the client names, and the nonce, drawn when the stream is + # first written to, marks that life of the stream. A stream dropped and written to again gets a new nonce, so the id a client + # still holds from the dropped one matches nothing in the new one and is refused rather than matched to a newer event; + # and nothing in an id counts the events of other streams. + def store_event(stream_id, message) + stream_id = stream_id.to_s + bytes = message.nil? ? 0 : message.bytesize + + @mutex.synchronize do + stream = @streams.delete(stream_id) || { nonce: SecureRandom.hex(8), sequence: 0, bytes: 0, events: [] } + stream[:sequence] += 1 + event_id = "#{stream_id}/#{stream[:nonce]}-#{stream[:sequence]}" + stream[:events] << [event_id, message].freeze + stream[:bytes] += bytes + @total_bytes += bytes + + while stream[:events].size > @max_events_per_stream + _dropped_id, dropped_message = stream[:events].shift + dropped_bytes = dropped_message.nil? ? 0 : dropped_message.bytesize + stream[:bytes] -= dropped_bytes + @total_bytes -= dropped_bytes + end + + @streams[stream_id] = stream + drop_oldest_streams + + event_id + end + end + + def events_after(last_event_id) + return unless last_event_id.is_a?(String) + + stream_id, separator, _sequence = last_event_id.rpartition("/") + return if separator.empty? + + @mutex.synchronize do + stream = @streams[stream_id] + return unless stream + + index = stream[:events].index { |event_id, _message| event_id == last_event_id } + return unless index + + events = stream[:events].drop(index + 1).reject { |_event_id, message| message.nil? } + + [stream_id, events] + end + end + + private + + # The stream just written is never dropped: a message larger than `max_bytes` on its own would otherwise + # evict the stream it was stored on and make its own event id unusable at once. + def drop_oldest_streams + while @streams.size > 1 && (@streams.size > @max_streams || @total_bytes > @max_bytes) + _dropped_id, dropped = @streams.shift + + @total_bytes -= dropped[:bytes] + end + end +end diff --git a/lib/mcp/server/transports.rb b/lib/mcp/server/transports.rb index 1c4bc1c3..7ed70a0a 100644 --- a/lib/mcp/server/transports.rb +++ b/lib/mcp/server/transports.rb @@ -3,6 +3,7 @@ module MCP class Server module Transports + autoload :EventStore, "mcp/server/transports/event_store" autoload :StdioTransport, "mcp/server/transports/stdio_transport" autoload :StreamableHTTPTransport, "mcp/server/transports/streamable_http_transport" end diff --git a/lib/mcp/server/transports/event_store.rb b/lib/mcp/server/transports/event_store.rb new file mode 100644 index 00000000..49201c1c --- /dev/null +++ b/lib/mcp/server/transports/event_store.rb @@ -0,0 +1,34 @@ +# frozen_string_literal: true + +module MCP + class Server + module Transports + # The contract of the store `StreamableHTTPTransport` keeps SSE events in for resumability (MCP 2025-11-25, + # SEP-1699): the transport stores each event it writes under the id the store assigns, and a client that + # reconnects with `Last-Event-ID` is sent the events of that stream stored after the one it names. + # + # A store serves every session of the transport. Stream ids are opaque strings the transport mints, + # one for the GET stream of a session and one for the response stream of each request; a store orders + # the events of one stream by storage order and never has to relate streams to each other. + # The SDK ships this contract and no store of its own. `examples/in_memory_event_store.rb` in its repository + # is one that keeps events in the process, to copy into an application; a deployment whose reconnects may reach + # another process implements this interface against a shared backend. + class EventStore + # Stores `message`, the JSON text of a JSON-RPC message, on the stream `stream_id` and returns the id + # assigned to it, a String without line breaks that is unique across the store. `message` is `nil` for + # the priming event that opens a stream, which is stored so that its id is a cursor like any other, + # but is never replayed. + def store_event(stream_id, message) + raise NotImplementedError, "#{self.class} must implement #store_event" + end + + # The events stored on the same stream after the event `last_event_id`, as `[stream_id, events]` + # where `events` is an Array of `[event_id, message]` pairs in storage order, priming events left out; + # `nil` when the store does not know `last_event_id`, which includes an event it has since dropped. + def events_after(last_event_id) + raise NotImplementedError, "#{self.class} must implement #events_after" + end + end + end + end +end diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index e3b7a1b4..15cc3dc8 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -218,6 +218,13 @@ class InvalidJsonError < StandardError; end # expiry closes the stream earlier when it comes first; the cap is what bounds a stream whose token reports no # expiry. `nil` removes the cap, leaving such a stream open until either side disconnects. Streams opened without # a token are never capped. + # @param event_store [MCP::Server::Transports::EventStore, nil] enables SSE stream resumability (MCP 2025-11-25, + # SEP-1699): every event is stored under the id the store assigns and written with it, each stream opens with + # a priming event, and a GET carrying `Last-Event-ID` replays the events of that stream the client missed and + # resumes it. The SDK ships no store; `examples/in_memory_event_store.rb` in its repository is one for + # a single process. Not supported in stateless mode, which has no sessions to resume. + # @param retry_interval [Integer, nil] milliseconds the SSE `retry` field of each priming event asks the client to + # wait before reconnecting; without it the client applies its own delay. Requires `event_store`. def initialize( server, stateless: false, @@ -241,7 +248,9 @@ def initialize( resource_metadata: nil, resource_metadata_url: nil, scope_matcher: nil, - max_stream_lifetime: DEFAULT_MAX_STREAM_LIFETIME + max_stream_lifetime: DEFAULT_MAX_STREAM_LIFETIME, + event_store: nil, + retry_interval: nil ) super(server) # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock, origin: origin_header }`. @@ -365,6 +374,27 @@ def initialize( @max_stream_lifetime = max_stream_lifetime + if event_store + raise ArgumentError, "event_store is not supported in stateless mode." if @stateless + + unless event_store.respond_to?(:store_event) && event_store.respond_to?(:events_after) + raise ArgumentError, "event_store must respond to store_event and events_after (see MCP::Server::Transports::EventStore)" + end + end + + @event_store = event_store + + if retry_interval + # A retry interval without a store would promise a resumable stream the transport cannot resume. + raise ArgumentError, "retry_interval requires event_store" unless event_store + + unless retry_interval.is_a?(Integer) && retry_interval.positive? + raise ArgumentError, "retry_interval must be a positive Integer of milliseconds or nil" + end + end + + @retry_interval = retry_interval + # `close` sets the flag and signals the reaper under `@mutex`, so the reaper leaves its wait at once instead of # being killed in the middle of a reap. @closing = false @@ -384,6 +414,17 @@ class StreamWriteTimeoutError < IOError; end # A write blocks only once the peer's socket buffers are full, which takes a reader already hundreds of kilobytes behind; # one that still has not taken the frame after this long is dropped whether it stopped reading or is only that slow. DEFAULT_STREAM_WRITE_TIMEOUT = 10 + + # The first protocol version whose clients parse an SSE event with empty data, which the priming event of + # SEP-1699 resumability is. A session that negotiated an earlier version gets no priming event and is never + # asked to resume, since it would have no event id to come back with; its other events still carry ids. + RESUMABLE_PROTOCOL_VERSION = "2025-11-25" + + # Replay passes a resume makes before it gives up attaching. Each pass takes the events stored since + # the previous one and a further pass follows whenever that batch was not empty, so a sender that stores + # at least one event during each write keeps the loop going; after this many passes that each found events, + # the connection is closed behind the last batch for the client to reconnect from. + REPLAY_PASS_LIMIT = 16 SESSION_REAP_INTERVAL = 60 # Seconds `close` waits for a reap in progress to finish closing its streams before it kills the reaper. REAPER_JOIN_TIMEOUT = 5 @@ -602,44 +643,41 @@ def deliver_targeted_notification(notification, session_id, related_request_id) return false if @enable_json_response && related_request_id streams_to_close = [] - - # Resolve the target stream under the lock, then write outside it: a stalled SSE reader - # must not block every other session that needs `@mutex`. - stream = @mutex.synchronize do - next unless (session = @sessions[session_id]) + live = @mutex.synchronize do + next false unless (session = @sessions[session_id]) if session_expired?(session) cleanup_and_collect_stream(session_id, streams_to_close) - next + next false end - active_stream(session, streams_to_close, related_request_id: related_request_id) + true end close_streams(streams_to_close) - return false unless stream + return false unless live - write_notification(stream, notification, session_id, related_request_id) + store_and_write(session_id, related_request_id, notification.to_json) end def deliver_broadcast_notification(notification) streams_to_close = [] - # Snapshot the connected streams under the lock, then write to each outside it. + # Snapshot the sessions under the lock, then write to each outside it. A session whose GET stream is + # closed at the moment still takes the notification when it has opened that stream before and an event + # store keeps it for the reconnect; one that never opened the stream has no cursor to resume from. targets = @mutex.synchronize do expired_session_ids = [] collected = @sessions.filter_map do |session_id, session| - next unless session[:get_sse_stream] + next unless session[:get_sse_stream] || (@event_store && session[:get_stream_opened]) if session_expired?(session) expired_session_ids << session_id next end - stream = active_stream(session, streams_to_close) - - [session_id, stream] if stream + session_id end expired_session_ids.each do |session_id| @@ -651,22 +689,12 @@ def deliver_broadcast_notification(notification) close_streams(streams_to_close) - targets.count do |session_id, stream| - write_notification(stream, notification, session_id, nil) + json = notification.to_json + targets.count do |session_id| + store_and_write(session_id, nil, json) end end - # Writes a notification to an SSE stream without holding `@mutex`. On a write error, - # drops the broken stream and returns false; on success returns true. - def write_notification(stream, notification, session_id, related_request_id) - push_to_stream(stream, notification) - true - rescue *STREAM_WRITE_ERRORS => e - MCP.configuration.exception_reporter.call(e, { session_id: session_id, error: "Failed to send notification" }) - drop_broken_stream(session_id, stream, related_request_id) - false - end - # Removes a stream that failed to accept a write. A request-scoped stream is dropped on its own; # a session-scoped (GET SSE) failure tears down the whole session. The `@sessions` mutation runs # under `@mutex`, and the affected streams are closed outside it. @@ -675,6 +703,8 @@ def write_notification(stream, notification, session_id, related_request_id) # the lock that removes the session: the failed write may have been picked for a stream that was detached since, # because its token expired and the client reopened, and the stream opened in its place must stay. A failed # stream the session no longer holds was closed by whichever path detached it, so there is nothing left to do. + # With an event store the GET stream is detached instead and the session kept: the client resumes the stream + # from its last event id (SEP-1699), and the events it misses meanwhile are stored for it. def drop_broken_stream(session_id, stream, related_request_id) streams_to_close = [] @@ -689,7 +719,11 @@ def drop_broken_stream(session_id, stream, related_request_id) streams_to_close << stream elsif session && session[:get_sse_stream].equal?(stream) - cleanup_and_collect_stream(session_id, streams_to_close) + if @event_store + streams_to_close << detach_get_stream(session) + else + cleanup_and_collect_stream(session_id, streams_to_close) + end end end @@ -744,39 +778,20 @@ def send_request(method, params = nil, session_id: nil, related_request_id: nil, request = { jsonrpc: "2.0", id: request_id, method: method } request[:params] = params if params - # Register the pending response and resolve the stream under the lock, but perform - # the write outside it so a stalled reader cannot block every other session on `@mutex`. - streams_to_close = [] - stream = @mutex.synchronize do - unless (session = @sessions[session_id]) + # Register the pending response under the lock before the request is written, so a response arriving at once finds its entry; + # the write itself runs outside `@mutex`. + @mutex.synchronize do + unless @sessions.key?(session_id) raise "Session not found: #{session_id}." end @pending_responses[request_id] = { queue: pending_response, session_id: session_id } - - active_stream(session, streams_to_close, related_request_id: related_request_id) - end - - close_streams(streams_to_close) - - sent = false - if stream - begin - push_to_stream(stream, request) - sent = true - rescue *STREAM_WRITE_ERRORS - drop_broken_stream(session_id, stream, related_request_id) - end end - # TODO: Replace with event store + replay when resumability is implemented. - # Resumability is a separate MCP specification feature (SSE event IDs, Last-Event-ID replay, - # event store management) independent of sampling. - # See: https://modelcontextprotocol.io/specification/latest/basic/transports#resumability-and-redelivery - # - # The TypeScript and Python SDKs buffer messages and replay on reconnect. - # Until then, raise to prevent queue.pop from blocking indefinitely. - unless sent + # With an event store a request whose stream is disconnected is kept for replay, so the wait below can still + # be answered once the client resumes the stream (SEP-1699); without one there is nothing the client could + # come back to, so the wait is not started. + unless store_and_write(session_id, related_request_id, request.to_json) raise "No active stream for #{method} request." end @@ -857,6 +872,47 @@ def cancel_pending_request(request_id, reason: nil) end end + # Closes the SSE connection of the request `request_id` without ending the request (SEP-1699), so the client reconnects with + # `Last-Event-ID` after the `retry_interval` and the handler's remaining messages, kept by the event store meanwhile, + # reach it through the resumed connection. Returns `false` when there is nothing to close: no `event_store:`, + # a client that negotiated a version before 2025-11-25 (it received no priming event to come back with), + # a connection whose priming event the store failed to take (the client holds no cursor for it either), + # an unknown session or request, or a connection already closed. Handlers reach this through `server_context.close_sse_stream`. + # @api private + def close_sse_stream(session_id, request_id) + stream = @mutex.synchronize do + session = session_id && @sessions[session_id] + next unless session && resumable_session?(session) + + entry = session.dig(:post_stream_entries, request_id) + current = session.dig(:post_request_streams, request_id) + next unless entry && current && entry[:primed_stream].equal?(current) + + session[:post_request_streams].delete(request_id) + end + return false unless stream + + close_stream_safely(stream) + + true + end + + # Closes the session's standalone GET stream the same way, so the client polls for server-initiated messages. + # @api private + def close_standalone_sse_stream(session_id) + stream = @mutex.synchronize do + session = session_id && @sessions[session_id] + next unless session && resumable_session?(session) + next unless session[:get_sse_stream] && session[:get_primed_stream].equal?(session[:get_sse_stream]) + + detach_get_stream(session) + end + return false unless stream + + close_stream_safely(stream) + true + end + private # The wait between reaps is a condition variable wait under `@mutex` rather than a sleep, so `close` can end it @@ -895,9 +951,24 @@ def reap_expired_sessions end end - def send_to_stream(stream, data) + # One `write` per frame, so the frames of concurrent writers cannot interleave in the middle of each other. + def send_to_stream(stream, data, event_id: nil) message = data.is_a?(String) ? data : data.to_json - stream.write("data: #{message}\n\n") + frame = event_id ? "id: #{event_id}\ndata: #{message}\n\n" : "data: #{message}\n\n" + + stream.write(frame) + stream.flush + end + + # The priming event of SEP-1699: an id for the client to resume from, the `retry` field when configured, and + # empty data. The `data:` line is what makes the event one the client dispatches (and so sees the id of); + # an id without it would only be noted. + def send_priming_event_to_stream(stream, event_id) + fields = ["id: #{event_id}"] + fields << "retry: #{@retry_interval}" if @retry_interval + fields << "data:" + + stream.write("#{fields.join("\n")}\n\n") stream.flush end @@ -920,8 +991,8 @@ def send_ping_to_stream(stream) # # The deadline interrupts the write with the exception `Timeout` raises on its own, which no `rescue` of an `IOError` or a `StandardError` inside # a stream implementation catches, and becomes `StreamWriteTimeoutError` only here, where it leaves the write. - def push_to_stream(stream, data, subscription: nil) - within_stream_write_timeout(stream, subscription) { send_to_stream(stream, data) } + def push_to_stream(stream, data, event_id: nil, subscription: nil) + within_stream_write_timeout(stream, subscription) { send_to_stream(stream, data, event_id: event_id) } end def push_ping_to_stream(stream, subscription: nil) @@ -1900,6 +1971,14 @@ def handle_get(request) protocol_version_error = validate_protocol_version_header(request) return protocol_version_error if protocol_version_error + # A reconnect naming the last event the client received resumes that stream (SEP-1699). Without an event store the header is ignored and + # a plain stream is opened, since nothing was stored to replay. + last_event_id = header_bytes(request, "HTTP_LAST_EVENT_ID") + + if @event_store && last_event_id && !last_event_id.empty? + return resume_sse_stream(session_id, last_event_id, request.env[OAuth::ENV_KEY]) + end + return session_already_connected_response if live_get_stream?(session_id) setup_sse_stream(session_id, request.env[OAuth::ENV_KEY]) @@ -2302,6 +2381,12 @@ def handle_initialization(request, body_string, body) auth_issuer: auth_info&.issuer, auth_subject: auth_info&.subject, auth_client_id: auth_info&.client_id, + # Orders the session's event storage against a `Last-Event-ID` resume; see `store_and_write`. + store_mutex: Mutex.new, + # The stream id the events of the session's GET stream are stored under, fixed for the session's life + # so a client resuming that stream finds everything it missed under one cursor however often + # the connection was reopened. The response stream of each request gets its own id in `post_stream_entries`. + get_stream_id: @event_store && mint_stream_id(session_id), } true end @@ -2395,39 +2480,65 @@ def handle_regular_request(body_string, session_id, related_request_id: nil, aut # https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#sending-messages-to-the-server def handle_request_with_sse_response(body_string, session_id, server_session, related_request_id: nil, auth_info: nil) body = proc do |stream| - @mutex.synchronize do + # The stream id is kept with the request while it is in flight: it marks the request's events in + # the event store and routes a `Last-Event-ID` resume back to the request (SEP-1699). + # It is minted whether or not a store is configured, so the registration has one shape. + stream_id = mint_stream_id(session_id) + claimed = @mutex.synchronize do session = @sessions[session_id] - if session && related_request_id - session[:post_request_streams] ||= {} + next false unless session && related_request_id - # Claim the id only while it is free. `handle_regular_request` already refused the colliding request, - # so reaching an occupied slot means a race got past that check; leaving the first stream in place keeps - # its messages going where they belong. - session[:post_request_streams][related_request_id] ||= stream - end + session[:post_request_streams] ||= {} + session[:post_stream_entries] ||= {} + + # Claim the id only while it is free. `handle_regular_request` already refused the colliding request, + # so reaching an occupied slot means a race got past that check; leaving the first stream in place keeps + # its messages going where they belong. + next false if session[:post_request_streams].key?(related_request_id) || session[:post_stream_entries].key?(related_request_id) + + session[:post_request_streams][related_request_id] = stream + session[:post_stream_entries][related_request_id] = { stream_id: stream_id, resumed: nil } + + true end + write_priming_event(session_id, related_request_id, stream) if claimed + begin response = dispatch_handle_json(body_string, server_session, auth_info: auth_info) - send_to_stream(stream, response) if response + if response + # Written through the registration rather than to `stream`: the response of a request whose connection was + # closed and resumed meanwhile belongs on the resumed one, and is stored for a client still to reconnect. + if claimed + store_and_write(session_id, related_request_id, response, pushed: false) + else + send_to_stream(stream, response) + end + end ensure - if related_request_id - @mutex.synchronize do - session = @sessions[session_id] - # Only retire our own registration: a request that never claimed the id, or one whose claim has - # already been replaced, must not unregister the stream that owns it. - registered = session&.dig(:post_request_streams, related_request_id) + resumed = nil - session[:post_request_streams].delete(related_request_id) if registered.equal?(stream) + if claimed + resumed = @mutex.synchronize do + session = @sessions[session_id] + entry = session&.dig(:post_stream_entries, related_request_id) + # Only retire our own registration: one whose claim has already been replaced must not unregister + # the stream that owns it. The live stream is this request's own when it is the one opened here or + # the one a resume put in its place. + next unless entry && entry[:stream_id] == stream_id + + session[:post_stream_entries].delete(related_request_id) + registered = session[:post_request_streams][related_request_id] + if registered.equal?(stream) || (entry[:resumed] && registered.equal?(entry[:resumed])) + session[:post_request_streams].delete(related_request_id) + end + entry[:resumed] end end - begin - stream.close - rescue StandardError - # Ignore close-related errors from already closed/broken streams. - end + close_stream_safely(resumed) if resumed + close_stream_safely(stream) end end @@ -2706,7 +2817,10 @@ def setup_sse_stream(session_id, auth_info = nil) def create_sse_body(session_id, auth_info = nil) proc do |stream| stored = store_stream_for_session(session_id, stream, auth_info) - start_keepalive_thread(session_id, stream) if stored + next unless stored + + write_priming_event(session_id, nil, stream) + start_keepalive_thread(session_id, stream) end end @@ -2719,6 +2833,9 @@ def store_stream_for_session(session_id, stream, auth_info = nil) session[:get_sse_stream] = stream session[:get_sse_stream_expires_at] = stream_token_expiry(auth_info) + # From here on the session's GET stream events are stored, since the client now has the cursor to + # resume them from. + session[:get_stream_opened] = true true end @@ -2737,7 +2854,8 @@ def store_stream_for_session(session_id, stream, auth_info = nil) # stream at a delivery once its token expired and the client may have opened a new one since, which gets # a keepalive thread of its own. The loop therefore ends as soon as the session no longer holds this stream, # and a loop that did not run to its end (a failed ping, or any other abnormal exit) drops the stream through - # `drop_broken_stream`, which tears the session down only while it still holds that stream, as before. + # `drop_broken_stream`, which tears the session down only while it still holds that stream, as before, + # or detaches the stream alone when an event store lets the client resume it. def start_keepalive_thread(session_id, stream) Thread.new do ended = false @@ -2820,6 +2938,307 @@ def session_expired?(session) Process.clock_gettime(Process::CLOCK_MONOTONIC) - session[:last_active_at] > @session_idle_timeout end + + # Stream ids carry the session id, so a `Last-Event-ID` is matched to the session presenting it with + # a string check, however long ago the request that opened the stream finished. + def mint_stream_id(session_id) + "#{session_id}:#{SecureRandom.uuid}" + end + + def owned_stream_id?(session_id, stream_id) + stream_id.is_a?(String) && stream_id.start_with?("#{session_id}:") + end + + def resumable_session?(session) + !@event_store.nil? && resumable_protocol_version?(session[:server_session].protocol_version) + end + + # Membership in the supported list keeps an unknown future version out: `initialize` only ever records + # a negotiated handshake version, so this is belt and braces, as in the TypeScript SDK. + def resumable_protocol_version?(version) + version.is_a?(String) && + MCP::Configuration::SUPPORTED_STABLE_PROTOCOL_VERSIONS.include?(version) && + version >= RESUMABLE_PROTOCOL_VERSION + end + + # The event store stream the messages for `related_request_id` belong to (the GET stream for `nil`), or `nil` + # when nothing is to be stored: no store, a request not in flight, or a GET stream the client has not opened + # yet and so has no cursor into. + def resolve_stream_id(session, related_request_id) + return unless @event_store + + if related_request_id + session.dig(:post_stream_entries, related_request_id, :stream_id) + elsif session[:get_stream_opened] + session[:get_stream_id] + end + end + + # Runs the block under the session's store lock, which orders the storing of a message and the choice of its + # stream against a replay pass and the attachment of a resumed stream (see `store_and_write`); `false` when + # the session is gone. The lock is created on first use, so a session built before this transport kept one is + # served too. It is held across store calls but never across a stream write, so a stalled reader stalls nobody. + def with_session_store_lock(session_id) + lock = @mutex.synchronize do + session = @sessions[session_id] + session && (session[:store_mutex] ||= Mutex.new) + end + return false unless lock + + lock.synchronize { yield } + true + end + + # Delivers `json`, a JSON-RPC message, on the session's stream for `related_request_id` (the GET stream for `nil`). + # With an event store the message is stored first and written with the id the store assigned, so one written while + # the stream is disconnected is replayed when the client resumes. The store and the choice of the stream share one hold of + # the session's store lock, which `replay_and_attach` holds across each replay pass and the attachment of the resumed stream: + # a message is therefore either in a set that replay writes to the resumed stream or written to that stream by + # its own sender afterwards, never missed by both or written to it twice. The write itself runs outside every lock. + # Returns `true` when the message reached a live stream or was stored for the reconnect, `false` otherwise. + # + # A message the transport pushes on its own (a notification, a server-to-client request) is written under `stream_write_timeout:`, + # since its sender writes to stream after stream. `pushed: false` is for the response of a request, + # written on that request's own thread, which waits for its reader as it did before. + def store_and_write(session_id, related_request_id, json, pushed: true) + event_id = nil + stream = nil + streams_to_close = [] + locked = with_session_store_lock(session_id) do + stream_id = @mutex.synchronize do + session = @sessions[session_id] + session && resolve_stream_id(session, related_request_id) + end + event_id = stream_id && safe_store_event(stream_id, json) + + stream = @mutex.synchronize do + session = @sessions[session_id] + session && active_stream(session, streams_to_close, related_request_id: related_request_id) + end + end + + close_streams(streams_to_close) + return false unless locked + + written = false + if stream + begin + if pushed + push_to_stream(stream, json, event_id: event_id) + else + send_to_stream(stream, json, event_id: event_id) + end + + written = true + rescue *STREAM_WRITE_ERRORS => e + MCP.configuration.exception_reporter.call(e, { session_id: session_id, error: "Failed to send message" }) + + drop_broken_stream(session_id, stream, related_request_id) + end + end + + written || !event_id.nil? + end + + # Opens `stream`, the connection the caller has just put in place for `related_request_id`, with the priming event, + # stored first so its id is a cursor in the stream like any other. Only for sessions that negotiated + # `RESUMABLE_PROTOCOL_VERSION` or later. + # + # The write has no deadline, since it is made on the thread of the request that opened or resumed the connection, + # so it goes to that connection and no other: when a later resume has taken its place in the meantime, + # nothing is written, and that resume primes the connection it attached. + def write_priming_event(session_id, related_request_id, stream) + event_id = nil + streams_to_close = [] + with_session_store_lock(session_id) do + stream_id = @mutex.synchronize do + session = @sessions[session_id] + next unless session && resumable_session?(session) + next unless active_stream(session, streams_to_close, related_request_id: related_request_id).equal?(stream) + + resolve_stream_id(session, related_request_id) + end + event_id = stream_id && safe_store_event(stream_id, nil) + end + close_streams(streams_to_close) + return unless event_id + + send_priming_event_to_stream(stream, event_id) + mark_primed(session_id, related_request_id, stream) + rescue *STREAM_WRITE_ERRORS => e + MCP.configuration.exception_reporter.call(e, { session_id: session_id, error: "Failed to send message" }) + drop_broken_stream(session_id, stream, related_request_id) + end + + # Records that `stream` received its priming event, which `close_sse_stream` and `close_standalone_sse_stream` + # require before they close it on purpose: a connection without one has given the client no cursor to come + # back with. Compared by identity, so a connection opened in the stream's place starts unprimed. + def mark_primed(session_id, related_request_id, stream) + @mutex.synchronize do + session = @sessions[session_id] + next unless session + + if related_request_id + entry = session.dig(:post_stream_entries, related_request_id) + entry[:primed_stream] = stream if entry && session.dig(:post_request_streams, related_request_id).equal?(stream) + elsif session[:get_sse_stream].equal?(stream) + session[:get_primed_stream] = stream + end + end + end + + # A failing store must not take the request or the session down with it: the frame is written without an id + # (the client's cursor stays at the previous event, which costs it this one event on resume) and the failure + # is reported. An id with a line break would end the SSE frame early, so it is refused the same way. + def safe_store_event(stream_id, message) + event_id = @event_store.store_event(stream_id, message) + return event_id if valid_event_id?(event_id) + + report_event_store_failure(ArgumentError.new("event store returned an invalid event id"), stream_id) + nil + rescue StandardError => e + report_event_store_failure(e, stream_id) + nil + end + + def safe_events_after(last_event_id) + replay = @event_store.events_after(last_event_id) + return unless replay.is_a?(Array) && replay.size == 2 && replay[0].is_a?(String) && replay[1].is_a?(Array) + + replay + rescue StandardError => e + report_event_store_failure(e, nil) + nil + end + + def valid_event_id?(event_id) + event_id.is_a?(String) && !event_id.empty? && !event_id.match?(/[\r\n]/) + end + + def report_event_store_failure(error, stream_id) + MCP.configuration.exception_reporter.call(error, { stream_id: stream_id, error: "Event store failure" }) + end + + # Serves a GET carrying `Last-Event-ID` (SEP-1699): the stream the event belongs to is replayed from that event and, + # when it is still open on the server side (the session's GET stream, or the response stream of a request still in flight), + # the new connection takes the old one's place. The stream must belong to this session: an id of another session, + # or one the store no longer knows, is answered 400 without replaying anything, and the two cases are not told apart. + def resume_sse_stream(session_id, last_event_id, auth_info) + replay = safe_events_after(last_event_id) + return unknown_last_event_id_response unless replay && owned_stream_id?(session_id, replay[0]) + + body = proc do |stream| + replay_and_attach(session_id, last_event_id, stream, auth_info) + end + + [200, SSE_HEADERS.dup, body] + end + + # The value is not echoed: it is client-supplied and nothing about it needs confirming. + def unknown_last_event_id_response + json_rpc_error_response( + status: 400, + code: JsonRpcHandler::ErrorCode::INVALID_REQUEST, + message: "Bad Request: Unknown Last-Event-ID", + ) + end + + # Writes the events stored on the stream after `last_event_id` to `stream`, then puts `stream` in place of the connection + # that stream was last carried by. Each pass takes the events stored since the cursor under the session's store lock + # and writes them outside it; the pass that finds none left attaches the stream under the same hold, so a message stored + # between the last replayed event and the attachment is impossible: it is either replayed by a further pass or written to + # the attached stream by its own sender. A resume whose `REPLAY_PASS_LIMIT` passes all found events is closed behind + # the last batch instead of attached, so no connection replays without end and the lock is still never held across a write; + # the client reconnects from its new cursor after `retry`, as after any disconnect. + def replay_and_attach(session_id, last_event_id, stream, auth_info) + cursor = last_event_id + attachment = nil + passes = 0 + + until attachment + events = [] + passes += 1 + locked = with_session_store_lock(session_id) do + replay = safe_events_after(cursor) + stream_id = replay && replay[0] + if stream_id && owned_stream_id?(session_id, stream_id) + events = replay[1] + attachment = attach_resumed_stream(session_id, stream_id, stream, auth_info) if events.empty? + else + attachment = :refused + end + end + attachment = :refused unless locked + attachment ||= :behind if passes >= REPLAY_PASS_LIMIT + + events.each do |event_id, message| + cursor = event_id + next unless message.is_a?(String) + + send_to_stream(stream, message, event_id: valid_event_id?(event_id) ? event_id : nil) + end + end + + finish_resumed_stream(session_id, stream, attachment) + rescue *STREAM_WRITE_ERRORS => e + MCP.configuration.exception_reporter.call(e, { session_id: session_id, error: "Failed to replay events" }) + + close_stream_safely(stream) + end + + # Puts `stream` in place of the connection `stream_id` was last carried by, the session's GET stream or + # the response stream of a request still in flight, and says which; `:retired` for the stream of a request already answered, + # `:gone` for a session that no longer exists. A connection still attached there is handed back as `replaced` for closing: + # the client presenting the id of its last event is the one that held it. Called under the session's store lock. + def attach_resumed_stream(session_id, stream_id, stream, auth_info) + @mutex.synchronize do + session = @sessions[session_id] + next :gone unless session + + if stream_id == session[:get_stream_id] + replaced = detach_get_stream(session) + session[:get_sse_stream] = stream + session[:get_sse_stream_expires_at] = stream_token_expiry(auth_info) + session[:get_stream_opened] = true + { kind: :get, replaced: replaced } + elsif (request_id = request_id_of_stream(session, stream_id)) + replaced = session[:post_request_streams][request_id] + session[:post_request_streams][request_id] = stream + session[:post_stream_entries][request_id][:resumed] = stream + { kind: :post, request_id: request_id, replaced: replaced } + else + :retired + end + end + end + + # The stream of a request already answered has nothing more to carry once replayed, so that connection is + # closed, as the spec asks after the response, and so is one that found events on each of its `REPLAY_PASS_LIMIT` + # passes, which the client reconnects from its new cursor. An attached stream is primed, giving the client a cursor and + # the `retry` field again, and a GET stream gets its keepalive. + def finish_resumed_stream(session_id, stream, attachment) + unless attachment.is_a?(Hash) + close_stream_safely(stream) + return + end + + close_stream_safely(attachment[:replaced]) if attachment[:replaced] + + if attachment[:kind] == :get + write_priming_event(session_id, nil, stream) + start_keepalive_thread(session_id, stream) + else + write_priming_event(session_id, attachment[:request_id], stream) + end + end + + def request_id_of_stream(session, stream_id) + entries = session[:post_stream_entries] + return unless entries + + request_id, = entries.find { |_request_id, entry| entry[:stream_id] == stream_id } + request_id + end end end end diff --git a/lib/mcp/server_context.rb b/lib/mcp/server_context.rb index c8d90e0d..52de4a2f 100644 --- a/lib/mcp/server_context.rb +++ b/lib/mcp/server_context.rb @@ -154,6 +154,25 @@ def notify_resources_updated(uri:) @notification_target.notify_resources_updated(uri: uri) end + # Closes the SSE connection carrying the current request's stream without ending the request (SEP-1699): + # the client reconnects with `Last-Event-ID` after the transport's `retry_interval`, and the messages + # the handler emits meanwhile are replayed to it. Use it to let a long-running operation go on without + # a connection held open for it. Returns `true` when a connection was closed and `false` when there was none to close: + # a transport without an event store, a client that negotiated a protocol version before 2025-11-25, + # a request answered as JSON, or a connection closed already. + def close_sse_stream + return false unless @related_request_id && @notification_target.respond_to?(:close_sse_stream) + + @notification_target.close_sse_stream(related_request_id: @related_request_id) + end + + # Closes the session's standalone GET stream the same way, so the client polls for server-initiated messages. + def close_standalone_sse_stream + return false unless @notification_target.respond_to?(:close_standalone_sse_stream) + + @notification_target.close_standalone_sse_stream + end + # Delegates to the session so the request is scoped to the originating client. # The originating request id is stamped as `related_request_id`, satisfying # SEP-2260's requirement that server-to-client requests be associated with diff --git a/lib/mcp/server_session.rb b/lib/mcp/server_session.rb index e9f0c02c..7b0f0110 100644 --- a/lib/mcp/server_session.rb +++ b/lib/mcp/server_session.rb @@ -323,6 +323,22 @@ def notify_log_message(data:, level:, logger: nil, related_request_id: nil) @server.configuration.exception_reporter.call(e, { notification: "log_message" }) end + # Closes the SSE connection of the request `related_request_id` without ending the request (SEP-1699), so + # the client polls the stream by reconnecting with `Last-Event-ID`; prefer `server_context.close_sse_stream` + # inside a handler, which names the current request. Returns `false` on a transport without resumable streams. + def close_sse_stream(related_request_id:) + return false unless @transport.respond_to?(:close_sse_stream) + + @transport.close_sse_stream(@session_id, related_request_id) + end + + # Closes the session's standalone GET stream the same way. + def close_standalone_sse_stream + return false unless @transport.respond_to?(:close_standalone_sse_stream) + + @transport.close_standalone_sse_stream(@session_id) + end + private def validate_era!(era) diff --git a/test/conformance/server_runner_test.rb b/test/conformance/server_runner_test.rb index a0ed7a09..23119631 100644 --- a/test/conformance/server_runner_test.rb +++ b/test/conformance/server_runner_test.rb @@ -111,6 +111,23 @@ class ServerRunnerTest < ActiveSupport::TestCase ) end + test "the referee runs with unhandled rejections downgraded to warnings, keeping the NODE_OPTIONS already set" do + original = ENV["NODE_OPTIONS"] + runner = ServerRunner.new + + ENV.delete("NODE_OPTIONS") + assert_equal({ "NODE_OPTIONS" => "--unhandled-rejections=warn" }, runner.send(:referee_env)) + + ENV["NODE_OPTIONS"] = "--max-old-space-size=512" + assert_equal({ "NODE_OPTIONS" => "--max-old-space-size=512 --unhandled-rejections=warn" }, runner.send(:referee_env)) + ensure + if original + ENV["NODE_OPTIONS"] = original + else + ENV.delete("NODE_OPTIONS") + end + end + test "build_command runs a frozen requirement set and keeps the baseline" do command = ServerRunner.new(requirements: "2026-07-28").send(:build_command) diff --git a/test/examples/in_memory_event_store_test.rb b/test/examples/in_memory_event_store_test.rb new file mode 100644 index 00000000..a5cc37f9 --- /dev/null +++ b/test/examples/in_memory_event_store_test.rb @@ -0,0 +1,97 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "../../examples/in_memory_event_store" + +# The store the SDK offers as an example is the one the conformance server and the resumability tests run on, +# so its bounds and its ids are checked like library code. +class InMemoryEventStoreTest < ActiveSupport::TestCase + test "assigns ordered ids within a stream and replays the events after a given one" do + store = InMemoryEventStore.new + priming = store.store_event("s", nil) + first = store.store_event("s", "{\"n\":1}") + second = store.store_event("s", "{\"n\":2}") + store.store_event("t", "{\"n\":3}") + + assert_equal ["s", [[first, "{\"n\":1}"], [second, "{\"n\":2}"]]], store.events_after(priming) + assert_equal ["s", [[second, "{\"n\":2}"]]], store.events_after(first) + assert_equal ["s", []], store.events_after(second) + assert_equal 3, [priming, first, second].uniq.size + end + + test "leaves priming events out of a replay" do + store = InMemoryEventStore.new + first = store.store_event("s", "{\"n\":1}") + store.store_event("s", nil) + second = store.store_event("s", "{\"n\":2}") + + assert_equal ["s", [[second, "{\"n\":2}"]]], store.events_after(first) + end + + test "does not know an id it never assigned" do + store = InMemoryEventStore.new + store.store_event("s", "{}") + + assert_nil store.events_after("s/99") + assert_nil store.events_after("other/1") + assert_nil store.events_after("no-separator") + assert_nil store.events_after(nil) + end + + test "keeps at most max_events_per_stream events of a stream, dropping the oldest" do + store = InMemoryEventStore.new(max_events_per_stream: 2) + first = store.store_event("s", "a") + second = store.store_event("s", "b") + third = store.store_event("s", "c") + + assert_nil store.events_after(first), "a dropped event is no longer a cursor" + assert_equal ["s", [[third, "c"]]], store.events_after(second) + end + + test "keeps at most max_streams streams, dropping the one written to least recently" do + store = InMemoryEventStore.new(max_streams: 2) + a = store.store_event("a", "1") + b = store.store_event("b", "2") + a_again = store.store_event("a", "3") + c = store.store_event("c", "4") + + assert_nil store.events_after(b) + assert_equal ["a", [[a_again, "3"]]], store.events_after(a) + assert_equal ["c", []], store.events_after(c) + end + + test "keeps the retained messages under max_bytes, dropping the streams written to least recently" do + store = InMemoryEventStore.new(max_bytes: 10) + a = store.store_event("a", "aaaa") + b = store.store_event("b", "bbbb") + c = store.store_event("c", "cccc") + + assert_nil store.events_after(a) + assert_equal ["b", []], store.events_after(b) + assert_equal ["c", []], store.events_after(c) + end + + test "never drops the stream just written to, however large its message" do + store = InMemoryEventStore.new(max_bytes: 4) + big = store.store_event("a", "x" * 10) + + assert_equal ["a", []], store.events_after(big) + end + + test "never reuses an id after a stream was dropped and written to again" do + store = InMemoryEventStore.new(max_streams: 1) + first = store.store_event("s", "1") + store.store_event("t", "2") + again = store.store_event("s", "3") + + refute_equal first, again + assert_nil store.events_after(first), "an id of the dropped stream is not a cursor into the rewritten one" + assert_equal ["s", []], store.events_after(again) + end + + test "rejects bounds that are not positive Integers" do + assert_raises(ArgumentError) { InMemoryEventStore.new(max_events_per_stream: 0) } + assert_raises(ArgumentError) { InMemoryEventStore.new(max_streams: nil) } + assert_raises(ArgumentError) { InMemoryEventStore.new(max_bytes: 1.5) } + end +end diff --git a/test/mcp/server/transports/streamable_http_resumability_test.rb b/test/mcp/server/transports/streamable_http_resumability_test.rb new file mode 100644 index 00000000..da43b67c --- /dev/null +++ b/test/mcp/server/transports/streamable_http_resumability_test.rb @@ -0,0 +1,640 @@ +# frozen_string_literal: true + +require "test_helper" +require "rack" +require_relative "../../../../examples/in_memory_event_store" + +module MCP + class Server + module Transports + # SEP-1699 resumability of the Streamable HTTP transport: event ids, priming events, closing a stream from a handler, + # and resuming a stream from `Last-Event-ID`. + class StreamableHTTPResumabilityTest < ActiveSupport::TestCase + include InitializeParamsTestHelper + + # A stream that buffers writes and remains readable after close. + class TestStream + attr_reader :string + + def initialize + @string = +"" + @closed = false + end + + def write(data) + raise IOError, "closed stream" if @closed + + @string << data + end + + def flush + end + + def close + @closed = true + end + + def closed? + @closed + end + end + + # An event store whose backend is unreachable. + class FailingEventStore < EventStore + def store_event(_stream_id, _message) + raise IOError, "store unreachable" + end + + def events_after(_last_event_id) + nil + end + end + + # An event store whose ids would end an SSE frame early. + class LineBreakingEventStore < EventStore + def store_event(stream_id, _message) + "#{stream_id}/1\ndata: injected" + end + + def events_after(_last_event_id) + nil + end + end + + # An event store that always holds one more event past the cursor, as a handler emitting without pause leaves it. + class EndlessEventStore < EventStore + def initialize + super + @sequence = 0 + end + + def store_event(stream_id, _message) + "#{stream_id}/#{@sequence += 1}" + end + + def events_after(last_event_id) + stream_id = last_event_id.rpartition("/").first + [stream_id, [["#{stream_id}/#{@sequence += 1}", { jsonrpc: "2.0", method: "notifications/progress" }.to_json]]] + end + end + + # A stream whose connection fails after `fail_after` writes. + class BreakingStream < TestStream + def initialize(fail_after:) + super() + @writes_left = fail_after + end + + def write(data) + @writes_left -= 1 + raise Errno::ECONNRESET, "peer went away" if @writes_left.negative? + + super + end + end + + setup do + @server = Server.new(name: "test_server", tools: [], prompts: [], resources: []) + @store = InMemoryEventStore.new + @transport = build_transport(event_store: @store, retry_interval: 250) + end + + teardown do + @transport.close + end + + test "event_store is refused in stateless mode" do + error = assert_raises(ArgumentError) do + StreamableHTTPTransport.new(@server, stateless: true, event_store: InMemoryEventStore.new) + end + + assert_match(/stateless/, error.message) + end + + test "retry_interval requires an event store and a positive Integer of milliseconds" do + assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, retry_interval: 100) } + assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, event_store: InMemoryEventStore.new, retry_interval: 0) } + assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, event_store: InMemoryEventStore.new, retry_interval: 1.5) } + end + + test "an event store must respond to store_event and events_after" do + assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, event_store: Object.new) } + end + + test "a POST response stream opens with a priming event and the response carries an event id" do + session_id = initialize_session + + events = sse_events(run_stream(post(session_id, { jsonrpc: "2.0", method: "ping", id: "1" }))) + + assert_equal 2, events.size + priming, response = events + assert_equal "", priming[:data] + assert_equal 250, priming[:retry] + refute_nil priming[:id] + assert_equal "1", JSON.parse(response[:data])["id"] + refute_nil response[:id] + refute_equal priming[:id], response[:id] + assert_nil response[:retry] + + # The priming event is a cursor into the stream: the response is what follows it. + _stream_id, replay = @store.events_after(priming[:id]) + assert_equal [[response[:id], response[:data]]], replay + end + + test "without an event store the stream carries no event ids" do + transport = build_transport + session_id = initialize_session(transport: transport) + + events = sse_events(run_stream(post(session_id, { jsonrpc: "2.0", method: "ping", id: "1" }, transport: transport))) + + assert_equal 1, events.size + assert_nil events.first[:id] + assert_equal "1", JSON.parse(events.first[:data])["id"] + ensure + transport&.close + end + + test "a client that negotiated a version before 2025-11-25 gets event ids but no priming event" do + closed = :unset + @server.define_tool(name: "closer") do |server_context:| + closed = server_context.close_sse_stream + Tool::Response.new([{ type: "text", text: "done" }]) + end + session_id = initialize_session(protocol_version: "2025-06-18") + + events = sse_events(run_stream(post(session_id, tool_call("closer", id: "1")))) + + assert_equal 1, events.size + assert_equal "1", JSON.parse(events.first[:data])["id"] + refute_nil events.first[:id] + assert_nil events.first[:retry] + refute closed + end + + test "close_sse_stream closes the connection and the response is replayed on Last-Event-ID" do + closed = nil + @server.define_tool(name: "closer") do |server_context:| + closed = server_context.close_sse_stream + Tool::Response.new([{ type: "text", text: "done" }]) + end + session_id = initialize_session + + stream = run_stream(post(session_id, tool_call("closer", id: "7"))) + events = sse_events(stream) + + assert closed + assert stream.closed? + assert_equal 1, events.size, "only the priming event reached the closed connection" + assert_equal "", events.first[:data] + + resumed = run_stream(resume(session_id, events.first[:id])) + replayed = sse_events(resumed) + + assert_equal 1, replayed.size + response = JSON.parse(replayed.first[:data]) + assert_equal "7", response["id"] + assert_equal "done", response.dig("result", "content", 0, "text") + refute_nil replayed.first[:id] + assert resumed.closed?, "a stream of an answered request is closed once replayed" + end + + test "a request still running after close_sse_stream continues on the resumed connection" do + gate = Queue.new + @server.define_tool(name: "poller") do |server_context:| + server_context.close_sse_stream + gate.pop + server_context.report_progress(1, message: "after reconnect") + Tool::Response.new([{ type: "text", text: "done" }]) + end + session_id = initialize_session + + response = post(session_id, tool_call("poller", id: "9", progress_token: "tok")) + stream = TestStream.new + request_thread = Thread.new { response[2].call(stream) } + wait_until { stream.closed? } + priming_id = sse_events(stream).first[:id] + + resumed = run_stream(resume(session_id, priming_id)) + + refute resumed.closed?, "the stream of a request still in flight stays attached" + attached = sse_events(resumed) + assert_equal 1, attached.size + assert_equal "", attached.first[:data], "the resumed connection opens with a priming event" + assert_equal 250, attached.first[:retry] + + gate << :go + request_thread.join + + events = sse_events(resumed) + assert_equal 3, events.size + progress = JSON.parse(events[1][:data]) + assert_equal "notifications/progress", progress["method"] + assert_equal "after reconnect", progress.dig("params", "message") + assert_equal "9", JSON.parse(events[2][:data])["id"] + assert events.all? { |event| event[:id] }, "every event of the resumed connection carries an id" + assert resumed.closed?, "the connection is closed after the response" + assert_equal 1, sse_events(stream).size, "nothing more reached the closed connection" + end + + test "replay starts after the event the client names" do + @server.define_tool(name: "stepper") do |server_context:| + server_context.report_progress(1, message: "first") + server_context.close_sse_stream + server_context.report_progress(2, message: "second") + Tool::Response.new([{ type: "text", text: "done" }]) + end + session_id = initialize_session + + events = sse_events(run_stream(post(session_id, tool_call("stepper", id: "3", progress_token: "tok")))) + + assert_equal 2, events.size + assert_equal "first", JSON.parse(events[1][:data]).dig("params", "message") + + replayed = sse_events(run_stream(resume(session_id, events[1][:id]))) + + assert_equal ["second", nil], replayed.map { |event| JSON.parse(event[:data]).dig("params", "message") } + assert_equal "3", JSON.parse(replayed.last[:data])["id"] + end + + test "an unknown Last-Event-ID is refused without echoing the value" do + session_id = initialize_session + + response = resume(session_id, "no-such-stream/1") + + assert_equal 400, response[0] + body = JSON.parse(response[2][0]) + assert_equal JsonRpcHandler::ErrorCode::INVALID_REQUEST, body["error"]["code"] + assert_equal "Bad Request: Unknown Last-Event-ID", body["error"]["message"] + refute_includes response[2][0], "no-such-stream" + end + + test "an event id of another session is refused and nothing of that session is replayed" do + @server.define_tool(name: "closer") do |server_context:| + server_context.close_sse_stream + Tool::Response.new([{ type: "text", text: "secret" }]) + end + victim = initialize_session + other = initialize_session + priming_id = sse_events(run_stream(post(victim, tool_call("closer", id: "1")))).first[:id] + + response = resume(other, priming_id) + + assert_equal 400, response[0] + assert_equal "Bad Request: Unknown Last-Event-ID", JSON.parse(response[2][0]).dig("error", "message") + + # The session the stream belongs to can still resume it. + replayed = sse_events(run_stream(resume(victim, priming_id))) + assert_equal "secret", JSON.parse(replayed.first[:data]).dig("result", "content", 0, "text") + end + + test "a GET stream opens with a priming event and a notification sent while it is closed is replayed" do + session_id = initialize_session + first = run_stream(get(session_id)) + opening = sse_events(first) + + assert_equal 1, opening.size + assert_equal "", opening.first[:data] + assert_equal 250, opening.first[:retry] + + assert @transport.close_standalone_sse_stream(session_id) + assert first.closed? + + @server.notify_tools_list_changed + + second = run_stream(resume(session_id, opening.first[:id])) + events = sse_events(second) + + assert_equal 2, events.size + assert_equal "notifications/tools/list_changed", JSON.parse(events[0][:data])["method"] + assert_equal "", events[1][:data], "the resumed stream is primed again" + refute second.closed? + + # The resumed connection is the session's GET stream from here on. + @server.notify_prompts_list_changed + live = sse_events(second).last + assert_equal "notifications/prompts/list_changed", JSON.parse(live[:data])["method"] + refute_nil live[:id] + end + + test "a priming event is written only to the connection its request put in place" do + session_id = initialize_session + current = run_stream(get(session_id)) + assert_equal 1, sse_events(current).size + + # The state a resume finds when a later resume attached its own connection first: the session's GET stream + # is no longer the one this request holds. + superseded = TestStream.new + @transport.send(:write_priming_event, session_id, nil, superseded) + + assert_empty superseded.string + assert_equal 1, sse_events(current).size + end + + test "close_standalone_sse_stream reports false when the session holds no GET stream" do + session_id = initialize_session + + refute @transport.close_standalone_sse_stream(session_id) + refute @transport.close_standalone_sse_stream("no-such-session") + end + + test "a resume of the GET stream replaces a connection still attached" do + session_id = initialize_session + first = run_stream(get(session_id)) + priming_id = sse_events(first).first[:id] + + second = run_stream(resume(session_id, priming_id)) + + assert first.closed? + refute second.closed? + @server.notify_tools_list_changed + assert_equal 1, sse_events(first).size, "nothing more reaches the replaced connection" + assert_includes second.string, "notifications/tools/list_changed" + end + + test "events of a GET stream the client never opened are not stored" do + session_id = initialize_session + + @server.notify_tools_list_changed + + assert_empty @store.instance_variable_get(:@streams) + refute @transport.close_standalone_sse_stream(session_id) + end + + test "a store that fails to store an event is reported and the event is still written without an id" do + reported = [] + original_reporter = MCP.configuration.exception_reporter + MCP.configuration.exception_reporter = ->(error, context) { reported << [error, context] } + transport = build_transport(event_store: FailingEventStore.new) + session_id = initialize_session(transport: transport) + + events = sse_events(run_stream(post(session_id, { jsonrpc: "2.0", method: "ping", id: "1" }, transport: transport))) + + assert_equal 1, events.size, "no priming event without an id to give it" + assert_nil events.first[:id] + assert_equal "1", JSON.parse(events.first[:data])["id"] + assert_equal ["Event store failure"] * 2, reported.map { |_error, context| context[:error] } + assert reported.all? { |error, _context| error.is_a?(IOError) } + ensure + MCP.configuration.exception_reporter = original_reporter + transport&.close + end + + test "a store that returns an id holding a line break is reported and the event is written without an id" do + reported = [] + original_reporter = MCP.configuration.exception_reporter + MCP.configuration.exception_reporter = ->(error, context) { reported << [error, context] } + transport = build_transport(event_store: LineBreakingEventStore.new) + session_id = initialize_session(transport: transport) + + stream = run_stream(post(session_id, { jsonrpc: "2.0", method: "ping", id: "1" }, transport: transport)) + events = sse_events(stream) + + assert_equal 1, events.size + assert_nil events.first[:id] + refute_includes stream.string, "injected" + assert_equal ["Event store failure"] * 2, reported.map { |_error, context| context[:error] } + assert reported.all? { |error, _context| error.is_a?(ArgumentError) } + assert_equal ["event store returned an invalid event id"], reported.map { |error, _context| error.message }.uniq + ensure + MCP.configuration.exception_reporter = original_reporter + transport&.close + end + + test "a GET stream whose connection fails is detached and the session resumes it" do + session_id = initialize_session + response = get(session_id) + assert_equal 200, response[0] + broken = BreakingStream.new(fail_after: 1) + response[2].call(broken) + priming_id = sse_events(broken).first[:id] + refute_nil priming_id + + @server.notify_tools_list_changed + + assert broken.closed? + events = sse_events(run_stream(resume(session_id, priming_id))) + assert_equal "notifications/tools/list_changed", JSON.parse(events[0][:data])["method"] + assert_equal "", events[1][:data], "the resumed connection is the session's GET stream and is primed" + end + + test "a GET stream whose reader stalls is detached at the write deadline and the session resumes it" do + transport = build_transport(event_store: InMemoryEventStore.new, stream_write_timeout: 0.1) + session_id = initialize_session(transport: transport) + stalled = run_stream(get(session_id, transport: transport)) + priming_id = sse_events(stalled).first[:id] + refute_nil priming_id + # A peer that stopped reading: the write blocks until the transport's own deadline cuts it short. + stalled.define_singleton_method(:write) { |_data| sleep(30) } + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + transport.send_notification("notifications/tools/list_changed", nil, **{}) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_operator elapsed, :<, 5 + assert stalled.closed? + events = sse_events(run_stream(resume(session_id, priming_id, transport: transport))) + assert_equal "notifications/tools/list_changed", JSON.parse(events[0][:data])["method"] + ensure + transport&.close + end + + test "close_sse_stream reports false when the priming event could not be stored and the response stays on the connection" do + @server.define_tool(name: "closer") do |server_context:| + Tool::Response.new([{ type: "text", text: server_context.close_sse_stream.inspect }]) + end + original_reporter = MCP.configuration.exception_reporter + MCP.configuration.exception_reporter = ->(_error, _context) {} + transport = build_transport(event_store: FailingEventStore.new) + session_id = initialize_session(transport: transport) + + stream = run_stream(post(session_id, tool_call("closer", id: "1"), transport: transport)) + + assert_equal "false", messages(stream).first.dig("result", "content", 0, "text") + ensure + MCP.configuration.exception_reporter = original_reporter + transport&.close + end + + test "a resume whose replay passes keep finding events is closed after the batch of the last pass" do + transport = build_transport(event_store: EndlessEventStore.new) + session_id = initialize_session(transport: transport) + first = run_stream(get(session_id, transport: transport)) + priming_id = sse_events(first).first[:id] + + second = run_stream(resume(session_id, priming_id, transport: transport)) + + assert second.closed? + assert_equal StreamableHTTPTransport::REPLAY_PASS_LIMIT, messages(second).size + refute first.closed?, "the connection still attached stays the session's GET stream" + ensure + transport&.close + end + + test "a resume presenting another principal's token is refused before anything is replayed" do + alice = { OAuth::ENV_KEY => OAuth::AccessToken.new(token: "alice-token", subject: "alice", client_id: "client-a") } + bob = { OAuth::ENV_KEY => OAuth::AccessToken.new(token: "bob-token", subject: "bob", client_id: "client-b") } + session_id = initialize_session(env: alice) + priming_id = sse_events(run_stream(get(session_id, env: alice))).first[:id] + assert @transport.close_standalone_sse_stream(session_id) + @server.notify_tools_list_changed + + response = resume(session_id, priming_id, env: bob) + + assert_equal 404, response[0] + replayed = messages(run_stream(resume(session_id, priming_id, env: alice))) + assert_equal "notifications/tools/list_changed", replayed.first["method"] + end + + test "a server-to-client request sent on a closed connection reaches the resumed one and is answered" do + @server.define_tool(name: "pinger") do |server_context:| + server_context.close_sse_stream + result = server_context.ping + Tool::Response.new([{ type: "text", text: result.inspect }]) + end + session_id = initialize_session + + response = post(session_id, tool_call("pinger", id: "5")) + stream = TestStream.new + request_thread = Thread.new { response[2].call(stream) } + wait_until { stream.closed? } + priming_id = sse_events(stream).first[:id] + + resumed = run_stream(resume(session_id, priming_id)) + wait_until { resumed.string.include?("\"method\":\"ping\"") } + ping = messages(resumed).find { |message| message["method"] == "ping" } + refute_nil ping["id"] + + accepted = post(session_id, { jsonrpc: "2.0", id: ping["id"], result: {} }) + assert_equal 202, accepted[0] + request_thread.join + + final = messages(resumed).last + assert_equal "5", final["id"] + assert_equal "{}", final.dig("result", "content", 0, "text") + end + + test "close_sse_stream reports false on a request answered as JSON" do + closed = nil + @server.define_tool(name: "closer") do |server_context:| + closed = server_context.close_sse_stream + Tool::Response.new([{ type: "text", text: "done" }]) + end + transport = build_transport(event_store: InMemoryEventStore.new, enable_json_response: true) + session_id = initialize_session(transport: transport) + + response = post(session_id, tool_call("closer", id: "1"), transport: transport) + + assert_equal 200, response[0] + assert_equal "application/json", response[1]["content-type"] + refute closed + ensure + transport&.close + end + + test "ServerContext#close_sse_stream reports false on a session whose transport has no resumable streams" do + session = ServerSession.new(server: @server, transport: Object.new, session_id: "s") + context = ServerContext.new({}, progress: nil, notification_target: session, related_request_id: "1") + + refute context.close_sse_stream + refute context.close_standalone_sse_stream + refute ServerContext.new({}, progress: nil, notification_target: session).close_sse_stream + end + + private + + def build_transport(**options) + StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, **options) + end + + def initialize_session(transport: @transport, protocol_version: MCP::Configuration::LATEST_HANDSHAKE_PROTOCOL_VERSION, env: {}) + response = transport.handle_request(rack_request( + "POST", + env, + { jsonrpc: "2.0", method: "initialize", id: "init", params: initialize_params(protocolVersion: protocol_version) }.to_json, + )) + + assert_equal 200, response[0] + response[1]["mcp-session-id"] + end + + def tool_call(name, id:, progress_token: nil) + params = { name: name, arguments: {} } + params[:_meta] = { progressToken: progress_token } if progress_token + { jsonrpc: "2.0", id: id, method: "tools/call", params: params } + end + + def post(session_id, body, transport: @transport) + transport.handle_request(rack_request("POST", { "HTTP_MCP_SESSION_ID" => session_id }, body.to_json)) + end + + def get(session_id, transport: @transport, env: {}) + transport.handle_request(rack_request("GET", { "HTTP_MCP_SESSION_ID" => session_id }.merge(env))) + end + + def resume(session_id, last_event_id, transport: @transport, env: {}) + headers = { "HTTP_MCP_SESSION_ID" => session_id, "HTTP_LAST_EVENT_ID" => last_event_id }.merge(env) + transport.handle_request(rack_request("GET", headers)) + end + + def rack_request(method, headers, body = nil) + env = { + "REQUEST_METHOD" => method, + "PATH_INFO" => "/", + "CONTENT_TYPE" => "application/json", + "HTTP_ACCEPT" => method == "GET" ? "text/event-stream" : "application/json, text/event-stream", + "rack.input" => StringIO.new(body.to_s), + }.merge(headers) + + Rack::Request.new(env) + end + + # Runs the streaming body of an SSE response into a TestStream and returns the stream. + def run_stream(response) + assert_equal 200, response[0] + assert_equal "text/event-stream", response[1]["content-type"] + + stream = TestStream.new + response[2].call(stream) + stream + end + + # The SSE events written to `stream` so far, as `{ id:, retry:, data: }` with `data` the joined data lines + # (`""` for a priming event); comment frames are skipped. + def sse_events(stream) + stream.string.split("\n\n").filter_map do |frame| + event = { id: nil, retry: nil } + data_lines = [] + frame.each_line(chomp: true) do |line| + next if line.start_with?(":") + + name, value = line.split(":", 2) + value = value.to_s.sub(/\A /, "") + case name + when "id" then event[:id] = value + when "retry" then event[:retry] = Integer(value) + when "data" then data_lines << value + end + end + next if data_lines.empty? + + event.merge(data: data_lines.join("\n")) + end + end + + # The JSON-RPC messages among the events written to `stream`, priming events left out. + def messages(stream) + sse_events(stream).reject { |event| event[:data].empty? }.map { |event| JSON.parse(event[:data]) } + end + + def wait_until(timeout: 5) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + until yield + flunk("Timed out waiting for condition") if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + + sleep(0.01) + end + end + end + end + end +end