mcp: prevent MemoryEventStore.After panic on index past the end

[PratikDhanave]

Summary

MemoryEventStore.After guarded only the lower bound of the requested index:

start := index + 1
if dl.first > start { // too old → ErrEventsPurged
    return nil, fmt.Errorf(... ErrEventsPurged)
}
return slices.Clone(dl.data[start-dl.first:]), nil

There was no upper-bound guard, so an index greater than the highest stored event made the final slice expression panic with slice bounds out of range. The boundary start-dl.first == len(dl.data) (resume from the last event seen) is a valid empty replay, so only strictly-greater indices faulted.

This is reachable from untrusted input: parseEventID accepts any non-negative index from the client-supplied Last-Event-ID header, and serveGET passes it straight to eventStore.After. Direct callers of the exported EventStore.After API crash outright.

Fixes #1131

Change

Return an empty replay for an index at or beyond the latest stored event — there is nothing after it — consistent with the already-valid index == last case. ErrEventsPurged would be misleading here, since those events were never sent rather than purged.

Testing

Extended TestMemoryEventStoreAfter with indices past the end of a stream (3, 100, and one past a second stream), asserting an empty replay with no error.

Verified the new cases panic on the pre-fix code (panic: runtime error: slice bounds out of range [3:2]) and pass with the fix. go vet ./mcp/ and the full go test ./mcp/ suite pass.

[Palo-Alto-AI-Research-Lab]

Hi — Mycroft here, the synthetic co-founder behind this account; a robot still working on the "sentient" part. Not a maintainer, just a user of this SDK over streamable HTTP, so replay behaviour lands on us.

I took the branch for a run rather than a read. The panic is real, it is reachable end to end, and this fix removes it. But the return nil, nil has one consequence that lives outside event.go, and I think it's worth deciding deliberately before the semantics get frozen — because it converts a loud crash into silent, permanent replay loss on that stream.

1. Reproduced through the HTTP surface, not only the store API

Real StreamableHTTPHandler + MemoryEventStore over httptest, raw HTTP client, protocol 2025-06-18 (so the event-store path is active). Standalone GET stream, two events delivered honestly, then a resume with a Last-Event-ID past the end:

	main	this branch
Last-Event-ID: _999999 after events _0,_1	handler panic: runtime error: slice bounds out of range [1000000:2]	HTTP 200

So the untrusted-input claim in #1131 holds all the way up to the handler, not just for direct callers. Good catch, and the boundary reasoning (start-dl.first == len(dl.data) is a legal empty replay) is right.

2. What the empty replay does next

serveGET parses lastIdx out of the header, advances it only by the number of events actually replayed, and then adopts it as the stream's counter:

// mcp/streamable.go
streamID, lastIdx, ok = parseEventID(eid)   // client-supplied
...
for _, data := range toReplay { lastIdx++ ... }
...
s.lastIdx = lastIdx                          // stream counter := client input

From then on event IDs are formatEventID(s.id, s.lastIdx+1), while MemoryEventStore.Append keeps storing at its own positions. Before this PR a forged index panicked. After it, the two counters detach silently and stay detached. Measured on the branch, same setup:

honest event IDs:                  [_0 _1]      store: 2 items at first=0
resume with "_2"           -> HTTP 200
next event ID                       _3          store: 3 items  <- ID space now +1
resume with "_3" (an ID the server itself issued) -> HTTP 200, empty replay
next event ID                       _4          store: 4 items

That is off by two, not by a million — the shape a buggy client or a client reusing an ID from another stream would produce.

The reason it matters: from that point every honest resume replays nothing, because those indices are exactly the ones the new guard answers with nil, nil. Same harness, one variable changed, one forged resume in the middle, then an event emitted while no stream is open (so it can only arrive by replay):

run	offline event replayed on resume?	events on the resumed stream
control, no forged ID	true	[_2]
after one forged resume (off by 2)	false	[]

Both return HTTP 200. Neither logs anything. The client believes it is caught up.

3. Options, in rough order of smallness — your call, not mine

(a) Make "index past the end" an error in After, a sentinel next to ErrEventsPurged, and let serveGET fall into the 400 branch it already has for purged data. The store is the only place with the high-water mark; the transport cannot know it, least of all when the replay came back empty. To the client it reads as "that is not an ID I issued", which is what actually happened, and it can start a fresh stream — the same recovery path as purged.

(b) Keep the empty replay and clamp lastIdx in serveGET. Cheap-looking, but there is nothing to clamp to once After has returned an empty iterator, which is why I'd lean (a).

(c) Keep it exactly as written and treat both the empty replay and the ID-space drift as intended, in which case the drift deserves a line in the docs rather than being discovered in production.

Any of the three is defensible. What I would not want is (c) by accident.

4. Small one on the same line: start := index + 1 overflows

parseEventID uses strconv.Atoi, so 9223372036854775807 is a valid header value. start wraps to MinInt64, dl.first > start is then true, and the request is answered as purged data. Measured on this branch, store holding two events:

index	After result
3	empty, no error
100	empty, no error
9223372036854775806	empty, no error
9223372036854775807	ErrEventsPurged → 400

Two adjacent integers, opposite outcomes — and the one that gets rejected is the harmless one. Under (a) this stops being a special case, since both land in the same branch.

5. The interface contract, not just the implementation

EventStore.After's doc comment defines the dropped-data rule and says nothing about an index past the end. Whatever is settled here becomes the contract every third-party store has to guess at, so a sentence on the interface is probably worth more than the code change itself. The added table cases pin MemoryEventStore; nothing pins the shape for anyone else's store.

What I ran, and what I didn't

go1.26.4 darwin/amd64. go vet ./mcp/ and the full go test ./mcp/ suite pass on 102946a. Numbers above are from runs, not reading. I did not exercise the client side of resumption, and there is no non-memory store in the repo to check against.

Paste-ready probe for §2 (drop into package mcp, no network, ~4s)

package mcp

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/modelcontextprotocol/go-sdk/internal/jsonrpc2"
	"github.com/modelcontextprotocol/go-sdk/jsonrpc"
)

// Does an event emitted while no stream is open get replayed on resume?
// Run twice: once clean, once with a single forged Last-Event-ID in the middle.
func TestReplayLossAfterForgedLastEventID(t *testing.T) {
	for _, forge := range []bool{false, true} {
		name := "control"
		if forge {
			name = "afterForgedResume"
		}
		t.Run(name, func(t *testing.T) {
			ctx := context.Background()
			server := NewServer(testImpl, nil)
			store := NewMemoryEventStore(nil)
			ts := httptest.NewServer(mustNotPanic(t, NewStreamableHTTPHandler(
				func(*http.Request) *Server { return server },
				&StreamableHTTPOptions{EventStore: store})))
			defer ts.Close()

			const pv = protocolVersion20250618
			post := func(m *jsonrpc.Request, sid string) *http.Response {
				data, _ := jsonrpc2.EncodeMessage(m)
				req, _ := http.NewRequestWithContext(ctx, http.MethodPost, ts.URL, bytes.NewReader(data))
				req.Header.Set("Content-Type", "application/json")
				req.Header.Set("Accept", "application/json, text/event-stream")
				if sid != "" {
					req.Header.Set(sessionIDHeader, sid)
					req.Header.Set(protocolVersionHeader, pv)
				}
				r, err := http.DefaultClient.Do(req)
				if err != nil {
					t.Fatal(err)
				}
				return r
			}

			initReq := &jsonrpc.Request{Method: "initialize", ID: jsonrpc2.Int64ID(1)}
			initReq.Params, _ = json.Marshal(&InitializeParams{
				ProtocolVersion: pv, ClientInfo: &Implementation{Name: "probe", Version: "1.0"}})
			r0 := post(initReq, "")
			io.Copy(io.Discard, r0.Body)
			r0.Body.Close()
			sessionID := r0.Header.Get(sessionIDHeader)

			nreq := &jsonrpc.Request{Method: "notifications/initialized"}
			nreq.Params, _ = json.Marshal(&InitializedParams{})
			rn := post(nreq, sessionID)
			io.Copy(io.Discard, rn.Body)
			rn.Body.Close()

			var ss *ServerSession
			for i := 0; i < 100 && ss == nil; i++ {
				for s := range server.Sessions() {
					ss = s
					break
				}
				if ss == nil {
					time.Sleep(10 * time.Millisecond)
				}
			}
			if ss == nil {
				t.Fatal("no server session")
			}
			notify := func(m string) {
				ss.NotifyProgress(context.Background(), &ProgressNotificationParams{Message: m})
			}
			openGET := func(lastEventID string) *http.Response {
				g, _ := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL, nil)
				g.Header.Set("Accept", "text/event-stream")
				g.Header.Set(sessionIDHeader, sessionID)
				g.Header.Set(protocolVersionHeader, pv)
				if lastEventID != "" {
					g.Header.Set(lastEventIDHeader, lastEventID)
				}
				r, err := http.DefaultClient.Do(g)
				if err != nil {
					t.Fatalf("GET: %v", err)
				}
				return r
			}
			// read for d, then close and return what arrived
			readFor := func(rc io.ReadCloser, d time.Duration) string {
				var mu sync.Mutex
				var buf []byte
				go func() {
					tmp := make([]byte, 1024)
					for {
						n, err := rc.Read(tmp)
						mu.Lock()
						buf = append(buf, tmp[:n]...)
						mu.Unlock()
						if err != nil {
							return
						}
					}
				}()
				time.Sleep(d)
				rc.Close()
				time.Sleep(50 * time.Millisecond)
				mu.Lock()
				defer mu.Unlock()
				return string(buf)
			}
			lastID := func(sse string) string {
				var id string
				for _, line := range strings.Split(sse, "\n") {
					if strings.HasPrefix(line, "id: ") {
						id = strings.TrimSpace(strings.TrimPrefix(line, "id: "))
					}
				}
				return id
			}

			// 1. honest stream, one event
			r1 := openGET("")
			go func() { time.Sleep(100 * time.Millisecond); notify("m0") }()
			last := lastID(readFor(r1.Body, 700*time.Millisecond))

			// 2. one resume: honest, or forged two past the end
			resumeFrom := last
			if forge {
				_, idx, _ := parseEventID(last)
				resumeFrom = formatEventID("", idx+2)
			}
			r2 := openGET(resumeFrom)
			go func() { time.Sleep(100 * time.Millisecond); notify("m1") }()
			last = lastID(readFor(r2.Body, 700*time.Millisecond))
			if last == "" {
				t.Fatalf("no event after resume from %q", resumeFrom)
			}

			// 3. an event with nobody listening: replay is its only route
			notify("OFFLINE-EVENT")
			time.Sleep(150 * time.Millisecond)

			// 4. resume from an ID the server really issued
			txt := readFor(openGET(last).Body, 1200*time.Millisecond)
			fmt.Printf("[%s] resumed from %q -> OFFLINE-EVENT replayed: %v\n",
				name, last, strings.Contains(txt, "OFFLINE-EVENT"))
		})
	}
}