From 49903e9fe020649e1b455ef4eb94e83d4efe3b38 Mon Sep 17 00:00:00 2001 From: Koichi ITO Date: Tue, 3 Nov 2026 06:44:18 +0900 Subject: [PATCH] Resume SSE streams from `Last-Event-ID` in the Streamable HTTP transport ## Motivation and Context MCP 2025-11-25 (SEP-1699) lets a client whose SSE connection dropped reconnect with a GET carrying `Last-Event-ID` and receive what it missed, and lets a server close a connection on purpose during a long operation so the client polls instead of holding it open. The TypeScript, Python, Go, and Rust SDKs implement it with an event store on the server transport, a priming event carrying `retry`, and replay on reconnect. The Ruby SDK had the client side only, with a TODO in `send_request`: the referee's `server-sse-polling` scenario warned about the missing priming event and `retry` field and could not test the resume at all. `StreamableHTTPTransport` now takes an `event_store:`, with `retry_interval:` in milliseconds for the `retry` field. With one, every event of a session's GET stream and of each request's response stream is stored under the id the store assigns and written with it, each stream opens with a priming event, and a GET with `Last-Event-ID` replays what the client missed from the stream the event belongs to. When that request is still running, the new connection takes the place of the one that carried the stream, so the handler's remaining messages reach the client; the stream of a request already answered is closed once replayed, as the spec asks after the response. `server_context.close_sse_stream` closes the connection of the current request without ending it, the polling pattern the spec describes, and `close_standalone_sse_stream` does the same for the session's GET stream. `EventStore` documents the two methods a store implements; the SDK ships no store of its own. `examples/in_memory_event_store.rb` is one for a single process, bounded by events per stream, by streams, and by bytes, to copy into an application, as the reference SDKs keep theirs among their examples. Points where the design goes beyond the reference SDKs, and why: - Replay is bound to the session. Stream ids carry the session id, and a `Last-Event-ID` of another session is refused the same way as an unknown one: HTTP 400, nothing replayed, and the value not echoed. The TypeScript and Python SDKs rely on their event ids being unguessable; the check is explicit here because one store serves every session of the transport. - The priming event and the close hooks apply only to sessions that negotiated 2025-11-25 or later, as in the TypeScript and Python SDKs, since earlier clients do not parse an event with empty data. Their other events still carry ids. - A message is stored and its stream chosen under a per-session lock that a resume holds across each replay pass and the attachment of the new connection, so a message is either replayed or written to the attached connection, never both or neither. The lock is never held across a stream write, keeping the transport's rule that a stalled reader stalls nobody else. - A store that fails is reported and the event is written without an id, so an outage of the backend degrades resumability rather than failing requests; an id holding a line break is refused the same way, since it would end the SSE frame early. - Events of a session's GET stream are stored only once the client has opened that stream, which is when it has a cursor; a broadcast then reaches a session whose GET connection is closed at the moment through the store. `event_store:` is refused in stateless mode, which has no sessions to resume. The modern lifecycle (2026-07-28) has no resumable streams, as its spec says, and is untouched. - A GET stream whose write fails is detached from the session instead of ending it when a store is configured, since the client resumes the stream from its last event id; without a store the session is torn down as before. - `close_sse_stream` and `close_standalone_sse_stream` close a connection only once its priming event was stored and sent, so the client always holds a cursor to come back with; when the store failed it, they report `false` and the request is answered on the connection it has. - A resume whose replay passes keep finding events, since events keep being stored while each batch is written, is closed after its sixteenth batch instead of attached, and the client reconnects from its new cursor after `retry`; no connection replays without end, and the lock is still never held across a write. - The in-memory store marks each life of a stream with a random nonce in its event ids, so the id a client holds from a stream that was dropped and written to again matches nothing in the new one and a stale `Last-Event-ID` is refused rather than matched to a newer event; nothing in an id counts other streams' events. The conformance server configures the store with a 100 ms `retry`, and its `test_reconnection` tool closes its stream mid-call, as the TypeScript and Python fixtures do. The server runner now starts the referee with `--unhandled-rejections=warn`: 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, and that rejection ended the referee process mid-suite although it reports nothing. Node 22 and 24, the version CI runs, do not reject, and the referee's checks decide the result as before. A message the transport pushes on its own, a notification or a server-to-client request, is written under `stream_write_timeout:` as before, with its event id. With an event store a stream cut at that deadline is one more disconnected stream: the message is already stored, and the client receives it when it resumes. A failed write to a session's stream is reported as "Failed to send message", since requests travel that path as well as notifications; the `subscriptions/listen` paths keep "Failed to send notification". ## How Has This Been Tested? test/mcp/server/transports/streamable_http_resumability_test.rb, test/examples/in_memory_event_store_test.rb, and test/conformance/server_runner_test.rb. Before the change, `npx @modelcontextprotocol/conformance@alpha server --scenario server-sse-polling` warned on `server-sse-priming-event` and `server-sse-retry-field` and reported `server-sse-disconnect-resume` as untestable; it now passes 3/3 at the default and the 2025-11-25 wire. ## Breaking Changes None. Without `event_store:` the transport writes the same frames as before. --- conformance/expected_failures.yml | 4 - conformance/server.rb | 16 +- conformance/server_runner.rb | 18 +- docs/_server/transports.md | 66 ++ examples/README.md | 15 + examples/in_memory_event_store.rb | 102 +++ lib/mcp/server/transports.rb | 1 + lib/mcp/server/transports/event_store.rb | 34 + .../transports/streamable_http_transport.rb | 585 +++++++++++++--- lib/mcp/server_context.rb | 19 + lib/mcp/server_session.rb | 16 + test/conformance/server_runner_test.rb | 17 + test/examples/in_memory_event_store_test.rb | 97 +++ .../streamable_http_resumability_test.rb | 640 ++++++++++++++++++ 14 files changed, 1540 insertions(+), 90 deletions(-) create mode 100644 examples/in_memory_event_store.rb create mode 100644 lib/mcp/server/transports/event_store.rb create mode 100644 test/examples/in_memory_event_store_test.rb create mode 100644 test/mcp/server/transports/streamable_http_resumability_test.rb 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