diff --git a/docs/guide/mcp.md b/docs/guide/mcp.md
index db2d6f061..f03b9d835 100644
--- a/docs/guide/mcp.md
+++ b/docs/guide/mcp.md
@@ -143,6 +143,21 @@ ggcode handles these requests by routing them through the same `ask_user` intera
 
 No configuration is needed — elicitation is enabled automatically when an interactive session is active.
 
+## Subscription Streams (MCP 2026-07-28)
+
+Protocol revision 2026-07-28 added correlated notification streams: a client may open a subscription with the `subscriptions/listen` request, and every notification the server sends on that stream carries a `_meta` field binding it to the subscription. This closes a long-standing ambiguity — when an agent talks to several MCP servers concurrently, a bare `notifications/tools/list_changed` cannot be attributed to a specific connection with certainty.
+
+ggcode opens this stream automatically when a server connects, and self-detects protocol support:
+
+- The listen request asks for the exact change set ggcode cares about (tool/prompt/resource list changes, resource subscriptions). The server replies with a `notifications/subscriptions/acknowledged` control message confirming the granted subset.
+- If the server answers `-32601` (method not found), ggcode downgrades permanently for that connection — legacy servers never see a second `subscriptions/listen` call, and the classic uncorrelated notifications keep working exactly as before.
+- Cancelling a subscription (client shutdown or transport teardown) follows the spec by sending `notifications/cancelled` with the listen request id.
+- Per the protocol MUSTs, notifications with a subscription id for an unknown or unacknowledged subscription are dropped rather than forwarded, so a server bug cannot poison the client's cache invalidation logic.
+
+**Transport support**: stdio (streaming) today; the HTTP transport only probes for the feature and treats unsupported responses as a graceful downgrade. WebSocket servers currently keep using uncorrelated notifications.
+
+No configuration is needed — enablement is automatic and transparent, with fallback to the pre-2026-07-28 behavior on any server that predates the revision.
+
 ## Request Cancellation (Client-to-Server)
 
 When ggcode gives up on an in-flight MCP request — the user interrupts a tool call, or a request exceeds its deadline — it sends the server a `notifications/cancelled` notification referencing the outstanding request id (MCP spec 2025-03-26). Spec-compliant servers stop processing the cancelled request and free associated resources instead of finishing orphaned work (a long-running database query, a partial file upload, etc.).
diff --git a/internal/mcp/client.go b/internal/mcp/client.go
index e82da3f8d..fc675b88a 100644
--- a/internal/mcp/client.go
+++ b/internal/mcp/client.go
@@ -140,6 +140,18 @@ type Client struct {
 	// entries for the List*/ReadResource calls (see cacheable.go). Zero value
 	// is usable, so struct-literal clients (tests) stay safe.
 	listingCache listingCache
+
+	// Subscription-stream registry (MCP 2026-07-28 subscriptions/listen,
+	// see subscriptions.go). subs maps the normalized listen-request ID JSON
+	// to its Subscription, used by routeSubscriptionNotification for ack
+	// correlation and MUST-gating. Guarded by subMu. modernSub holds the
+	// default list-change stream opened by enableModernSubscriptions;
+	// subListenState caches the server's protocol support so a legacy server
+	// never sees a second subscriptions/listen call.
+	subMu          sync.Mutex
+	subs           map[string]*Subscription
+	modernSub      *Subscription
+	subListenState atomic.Int32
 }
 
 // negotiatedState returns the protocol version and server capabilities
@@ -710,6 +722,12 @@ func (c *Client) Abort() {
 		if c.notificationDone != nil {
 			close(c.notificationDone)
 		}
+
+		// End every open subscriptions/listen stream: the transport is gone,
+		// so no acknowledged notification or correlated traffic can arrive
+		// anymore (subscriptions.go). No notifications/cancelled is sent on
+		// this path — there is no connection left to deliver it on.
+		c.closeAllSubscriptions(fmt.Errorf("mcp[%s]: connection closed", c.name))
 	})
 }
 
@@ -2788,6 +2806,15 @@ func (c *Client) processNotification(notif *Notification) {
 		c.handleElicitationComplete(notif.Params)
 		return
 	}
+	// MCP 2026-07-28 subscriptions/listen: consume subscription control
+	// traffic and gate listen-stream notifications per the protocol MUSTs
+	// (unknown/unacked subscription ⇒ drop). Valid correlated traffic falls
+	// through so cache invalidation and the user handler still fire — the
+	// legacy dispatch is unchanged, this only adds correlation. Mutex-only
+	// work, safe on the read loop.
+	if c.routeSubscriptionNotification(notif) {
+		return
+	}
 	// MCP 2026-07-28 CacheableResult (SEP-2549): drop cached listings/read
 	// results affected by change notifications BEFORE the handler runs, so a
 	// hot refresh (ListTools etc.) observes fresh data. Mutex-only work —
diff --git a/internal/mcp/subscriptions.go b/internal/mcp/subscriptions.go
new file mode 100644
index 000000000..cc12ad272
--- /dev/null
+++ b/internal/mcp/subscriptions.go
@@ -0,0 +1,530 @@
+package mcp
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/topcheer/ggcode/internal/debug"
+	"github.com/topcheer/ggcode/internal/safego"
+)
+
+// Modern subscription streams — "subscriptions/listen" (MCP protocol
+// revision 2026-07-28, notifications/subscriptions/acknowledged).
+//
+// The initialize-era push model (standalone GET SSE stream on streamable
+// HTTP; plain stdio writes) carries no correlation metadata: a client cannot
+// prove a notification is addressed to it, and the server cannot scope what
+// it pushes. The 2026-07-28 revision replaces this with an explicit
+// request/response stream:
+//
+//	→ {"method": "subscriptions/listen",
+//	   "params": {"notifications": {"toolsListChanged": true, ...}}, "id": 1}
+//	← notifications/subscriptions/acknowledged {params._meta[subId], params.notifications}
+//	← ... correlated notifications, each params._meta carrying the
+//	    subscriptionId ...
+//	← {"result": {...}, "id": 1}          ← graceful stream closure
+//
+// Server-side MUSTs we implement:
+//   - the server MUST NOT deliver notifications before acknowledging, so any
+//     notification whose subscriptionId is unknown or not yet acknowledged is
+//     a protocol violation → dropped with a debug log;
+//   - every listen-stream notification carries params._meta[".../subscriptionId"]
+//     → used for correlation here, then stripped from the user's view (the
+//     legacy handler keeps receiving the plain method/params it always did);
+//   - notifications/subscriptions/acknowledged is a control message and is
+//     consumed here, never forwarded.
+//
+// Client-side MUST we implement: when closing a subscription before the
+// server does, send notifications/cancelled (reuses the existing
+// Client.notifyCancelled plumbing).
+//
+// Transport notes: stdio delivers the eventual response through the shared
+// read loop's waiter registry; streamable HTTP delivers it as the final SSE
+// event of the long-lived POST (interleaved notifications ride the same
+// stream into processNotification). WebSocket is intentionally unsupported
+// for listen (legacy ws bridge, no ggcode-managed server uses it).
+
+const (
+	// MethodSubscriptionsListen requests a correlated notification stream.
+	MethodSubscriptionsListen = "subscriptions/listen"
+	// NotificationSubscriptionsAcknowledged confirms the negotiated filter.
+	NotificationSubscriptionsAcknowledged = "notifications/subscriptions/acknowledged"
+	// MetaKeySubscriptionID is the params._meta key carrying the JSON-RPC id
+	// of the subscriptions/listen request a notification belongs to.
+	MetaKeySubscriptionID = "io.modelcontextprotocol/subscriptionId"
+
+	// mcpSubscriptionAckTimeout bounds the wait for the acknowledged
+	// notification. A compliant server acks before delivering anything, so a
+	// healthy roundtrip is milliseconds; anything slower is treated as a
+	// failed subscription and cancelled.
+	mcpSubscriptionAckTimeout = 15 * time.Second
+)
+
+// errCodeMethodNotFound is the JSON-RPC error a legacy server returns for an
+// unknown method; used to downgrade to the legacy push path permanently.
+const errCodeMethodNotFound = -32601
+
+// SubscriptionFilter describes which notification types a client requests on
+// a subscriptions/listen stream (params.notifications).
+type SubscriptionFilter struct {
+	ToolsListChanged      bool     `json:"toolsListChanged,omitempty"`
+	PromptsListChanged    bool     `json:"promptsListChanged,omitempty"`
+	ResourcesListChanged  bool     `json:"resourcesListChanged,omitempty"`
+	ResourceSubscriptions []string `json:"resourceSubscriptions,omitempty"`
+}
+
+func (f SubscriptionFilter) empty() bool {
+	return !f.ToolsListChanged && !f.PromptsListChanged && !f.ResourcesListChanged &&
+		len(f.ResourceSubscriptions) == 0
+}
+
+// missingFrom returns the human-readable entries requested by f that agreed
+// does not grant, for logging the ack diff.
+func (f SubscriptionFilter) missingFrom(agreed SubscriptionFilter) []string {
+	var missing []string
+	add := func(cond, granted bool, name string) {
+		if cond && !granted {
+			missing = append(missing, name)
+		}
+	}
+	add(f.ToolsListChanged, agreed.ToolsListChanged, "toolsListChanged")
+	add(f.PromptsListChanged, agreed.PromptsListChanged, "promptsListChanged")
+	add(f.ResourcesListChanged, agreed.ResourcesListChanged, "resourcesListChanged")
+	granted := make(map[string]bool, len(agreed.ResourceSubscriptions))
+	for _, uri := range agreed.ResourceSubscriptions {
+		granted[uri] = true
+	}
+	for _, uri := range f.ResourceSubscriptions {
+		if !granted[uri] {
+			missing = append(missing, "resourceSubscriptions:"+uri)
+		}
+	}
+	return missing
+}
+
+// Subscription is one open subscriptions/listen stream. The zero value is
+// not usable; instances come from Client.ListenSubscriptions.
+type Subscription struct {
+	client    *Client
+	id        *ID
+	requested SubscriptionFilter
+
+	cancelCtx  context.Context
+	cancelFunc context.CancelFunc
+
+	acked atomic.Bool
+	ackCh chan struct{} // closed exactly once on ack
+
+	mu     sync.Mutex // guards agreed + endErr
+	agreed SubscriptionFilter
+	endErr error
+
+	cancelled atomic.Bool
+	endOnce   sync.Once
+	done      chan struct{} // closed exactly once on stream end
+}
+
+// ID returns the normalized JSON form of the listen request id (the
+// subscriptionId servers echo back in _meta).
+func (s *Subscription) ID() string { return subscriptionIDKey(s.id) }
+
+// Done is closed when the stream ends, for any reason.
+func (s *Subscription) Done() <-chan struct{} { return s.done }
+
+// Err returns the terminal error, or nil for a graceful server closure.
+func (s *Subscription) Err() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.endErr
+}
+
+// Agreed returns the filter subset the server acknowledged.
+func (s *Subscription) Agreed() SubscriptionFilter {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.agreed
+}
+
+// Requested returns the filter that was originally requested.
+func (s *Subscription) Requested() SubscriptionFilter { return s.requested }
+
+// Cancel tears the subscription down: it sends notifications/cancelled on a
+// detached goroutine (mirroring the request-path cancellation contract),
+// unblocks the transport watch goroutine, and ends the stream. Idempotent.
+func (s *Subscription) Cancel(cause error) {
+	if !s.cancelled.CompareAndSwap(false, true) {
+		return
+	}
+	id := s.id
+	client := s.client
+	safego.Go("mcp.subs.cancelled", func() {
+		client.notifyCancelled(id, cause, MethodSubscriptionsListen)
+	})
+	s.end(nil)
+}
+
+// end terminates the stream bookkeeping. Called exactly once via endOnce;
+// also cancels the subscription-lifetime context so the transport watch
+// goroutine (blocked in the HTTP roundtrip or on the waiter channel) exits.
+func (s *Subscription) end(err error) {
+	s.endOnce.Do(func() {
+		if err != nil {
+			s.mu.Lock()
+			s.endErr = err
+			s.mu.Unlock()
+		}
+		s.cancelFunc()
+		close(s.done)
+		s.client.removeSubscription(subscriptionIDKey(s.id), s)
+	})
+}
+
+// markAcked records the acknowledged filter and releases the ack waiter.
+// Returns false if the subscription is already acked (duplicate acks are
+// idempotent per protocol tolerance of late duplicates).
+func (s *Subscription) markAcked(agreed SubscriptionFilter) {
+	if !s.acked.CompareAndSwap(false, true) {
+		return
+	}
+	s.mu.Lock()
+	s.agreed = agreed
+	s.mu.Unlock()
+	for _, name := range s.requested.missingFrom(agreed) {
+		debug.Log("mcp-subs", "server=%s subscription %s: server did not grant %s",
+			s.client.name, s.ID(), name)
+	}
+	close(s.ackCh)
+}
+
+// subscriptionIDKey normalizes a JSON-RPC id to its compact JSON form so a
+// numeric id like 7 in a request matches "7" echoed back in _meta
+// (json.Number/float ambiguity is removed by Compact).
+func subscriptionIDKey(id *ID) string {
+	if id == nil {
+		return ""
+	}
+	raw, err := json.Marshal(id)
+	if err != nil {
+		return ""
+	}
+	return string(raw)
+}
+
+func normalizeIDJSON(raw json.RawMessage) (string, bool) {
+	if len(raw) == 0 {
+		return "", false
+	}
+	var buf bytes.Buffer
+	if err := json.Compact(&buf, raw); err != nil {
+		return "", false
+	}
+	key := buf.String()
+	if key == "" || key == "null" {
+		return "", false
+	}
+	return key, true
+}
+
+// subState values for Client.subListenState.
+const (
+	subStateUnknown = iota
+	subStateSupported
+	subStateUnsupported
+)
+
+// EnableModernSubscriptions attempts to open the default list-change
+// subscription stream and records whether the server speaks the protocol at
+// all. A -32601 (method not found) response downgrades permanently for this
+// client instance — legacy servers must never see the call again, matching
+// the GET-stream 405 downgrade behavior for HTTP. Safe to call on every
+// (re)connect: an existing live subscription short-circuits.
+func (c *Client) EnableModernSubscriptions(ctx context.Context) bool {
+	if c.subListenState.Load() == subStateUnsupported {
+		return false
+	}
+	c.subMu.Lock()
+	existing := c.modernSub
+	c.subMu.Unlock()
+	if existing != nil {
+		select {
+		case <-existing.done:
+		default:
+			return true // already listening
+		}
+	}
+
+	filter := SubscriptionFilter{
+		ToolsListChanged:     true,
+		PromptsListChanged:   true,
+		ResourcesListChanged: true,
+	}
+	sub, err := c.ListenSubscriptions(ctx, filter)
+	if err != nil {
+		var rpcErr *Error
+		if errors.As(err, &rpcErr) && rpcErr.Code == errCodeMethodNotFound {
+			c.subListenState.Store(subStateUnsupported)
+			debug.Log("mcp-subs", "server=%s subscriptions/listen unsupported, using legacy push", c.name)
+			return false
+		}
+		debug.Log("mcp-subs", "server=%s subscriptions/listen failed: %v", c.name, err)
+		return false
+	}
+	c.subMu.Lock()
+	c.modernSub = sub
+	c.subMu.Unlock()
+	c.subListenState.Store(subStateSupported)
+	debug.Log("mcp-subs", "server=%s subscription %s active", c.name, sub.ID())
+	return true
+}
+
+// ListenSubscriptions opens one correlated notification stream with the
+// given filter and blocks until the server acknowledges it (or the wait
+// fails). The returned Subscription stays open after this call; its Done
+// channel reports the eventual graceful closure or terminal error. The
+// request deliberately bypasses mcpRequestTimeout: a compliant stream is
+// long-lived by design and its lifetime is governed by the subscription,
+// not the caller's ack-wait context.
+func (c *Client) ListenSubscriptions(ctx context.Context, filter SubscriptionFilter) (*Subscription, error) {
+	if c.closed.Load() {
+		return nil, fmt.Errorf("mcp[%s]: connection closed", c.name)
+	}
+	if filter.empty() {
+		return nil, fmt.Errorf("mcp[%s]: subscriptions/listen requires a non-empty filter", c.name)
+	}
+	transport := c.transport
+	switch transport {
+	case "", "stdio", "http":
+	default:
+		return nil, fmt.Errorf("mcp[%s]: subscriptions/listen unsupported on transport %q", c.name, transport)
+	}
+
+	reqID := c.nextRequestID()
+	params, err := json.Marshal(struct {
+		Notifications SubscriptionFilter `json:"notifications"`
+	}{Notifications: filter})
+	if err != nil {
+		return nil, fmt.Errorf("mcp[%s]: marshal subscriptions/listen params: %w", c.name, err)
+	}
+	req := Request{
+		JSONRPC: "2.0",
+		Method:  MethodSubscriptionsListen,
+		Params:  params,
+		ID:      reqID,
+	}
+
+	sub := &Subscription{
+		client:    c,
+		id:        reqID,
+		requested: filter,
+		ackCh:     make(chan struct{}),
+		done:      make(chan struct{}),
+	}
+	sub.cancelCtx, sub.cancelFunc = context.WithCancel(context.Background())
+
+	// stdio: register the waiter BEFORE the write (same ordering guarantee
+	// as #994) so the shared read loop can never drop the eventual response
+	// as unknown-ID traffic. HTTP: the watch goroutine owns the roundtrip
+	// directly — the response arrives as the terminal event of the POST.
+	var waiter chan *Response
+	if transport == "" || transport == "stdio" {
+		waiter = make(chan *Response, 1)
+		c.registerWaiter(reqID, waiter)
+		c.mu.Lock()
+		err = c.writeMessageUnlocked(req)
+		c.mu.Unlock()
+		if err != nil {
+			c.unregisterWaiter(reqID, waiter)
+			return nil, fmt.Errorf("mcp[%s]: write subscriptions/listen: %w", c.name, err)
+		}
+	}
+
+	c.addSubscription(subscriptionIDKey(reqID), sub)
+
+	safego.Go("mcp.subs.watch", func() {
+		defer sub.end(nil)
+		if waiter != nil {
+			defer c.unregisterWaiter(reqID, waiter)
+			select {
+			case resp, ok := <-waiter:
+				if !ok || resp == nil {
+					return
+				}
+				if resp.IsError() {
+					sub.end(resp.Error)
+				}
+				// Graceful closure ("resultType": "complete" or empty result).
+			case <-sub.cancelCtx.Done():
+			case <-c.notificationDone: // nil-safe: nil chan blocks forever; Close/Abort always cancels cancelCtx via closeAllSubscriptions
+			}
+			return
+		}
+		// HTTP: the long-lived POST returns when the server closes the
+		// stream gracefully, drops the connection, or the subscription is
+		// cancelled (cancelCtx aborts the in-flight roundtrip).
+		resp, err := c.sendHTTP(sub.cancelCtx, req)
+		select {
+		case <-sub.cancelCtx.Done():
+			return // cancelled by us; Cancel already handled the protocol side
+		default:
+		}
+		if err != nil {
+			sub.end(err)
+			return
+		}
+		if resp.IsError() {
+			sub.end(resp.Error)
+		}
+	})
+
+	select {
+	case <-sub.ackCh:
+		return sub, nil
+	case <-sub.done:
+		if err := sub.Err(); err != nil {
+			return nil, fmt.Errorf("mcp[%s]: subscriptions/listen: %w", c.name, err)
+		}
+		return nil, fmt.Errorf("mcp[%s]: subscriptions/listen ended before acknowledgement", c.name)
+	case <-ctx.Done():
+		sub.Cancel(ctx.Err())
+		return nil, fmt.Errorf("mcp[%s]: subscriptions/listen: %w", c.name, ctx.Err())
+	case <-time.After(mcpSubscriptionAckTimeout):
+		cause := fmt.Errorf("no %s within %s", NotificationSubscriptionsAcknowledged, mcpSubscriptionAckTimeout)
+		sub.Cancel(cause)
+		return nil, fmt.Errorf("mcp[%s]: subscriptions/listen: %w", c.name, cause)
+	}
+}
+
+// addSubscription registers sub in the correlation registry.
+func (c *Client) addSubscription(key string, sub *Subscription) {
+	c.subMu.Lock()
+	if c.subs == nil {
+		c.subs = make(map[string]*Subscription)
+	}
+	c.subs[key] = sub
+	c.subMu.Unlock()
+}
+
+// removeSubscription drops sub from the registry if it is still the entry.
+func (c *Client) removeSubscription(key string, sub *Subscription) {
+	c.subMu.Lock()
+	if c.subs != nil && c.subs[key] == sub {
+		delete(c.subs, key)
+	}
+	c.subMu.Unlock()
+}
+
+// closeAllSubscriptions ends every open subscription with cause. Called from
+// Abort (once) when the transport is torn down; no notifications/cancelled
+// is sent on this path — the connection is gone.
+func (c *Client) closeAllSubscriptions(cause error) {
+	c.subMu.Lock()
+	subs := make([]*Subscription, 0, len(c.subs))
+	for _, s := range c.subs {
+		subs = append(subs, s)
+	}
+	c.subs = nil
+	c.subMu.Unlock()
+	for _, s := range subs {
+		s.end(cause)
+	}
+}
+
+// routeSubscriptionNotification consumes subscription control traffic and
+// gates listen-stream notifications. Returns true when the notification has
+// been fully handled here and must NOT reach the legacy dispatch
+// (cacheInvalidateForNotification / user handler):
+//   - notifications/subscriptions/acknowledged: control message, always consumed;
+//   - any notification carrying _meta subscriptionId for an unknown or
+//     unacknowledged subscription: protocol violation, dropped + logged;
+//   - any notification carrying a subscriptionId for a live, acked
+//     subscription: correlation complete — but the notification itself is
+//     forwarded normally (cache invalidation + user handler must still fire).
+//
+// Notifications without the _meta key are legacy traffic and fall through.
+func (c *Client) routeSubscriptionNotification(notif *Notification) bool {
+	if notif.Method == NotificationSubscriptionsAcknowledged {
+		key, agreed, ok := parseSubscriptionAckParams(notif.Params)
+		if !ok {
+			debug.Log("mcp-subs", "server=%s malformed %s params dropped", c.name, notif.Method)
+			return true
+		}
+		c.subMu.Lock()
+		sub := c.subs[key]
+		c.subMu.Unlock()
+		if sub == nil {
+			debug.Log("mcp-subs", "server=%s %s for unknown subscription %s dropped",
+				c.name, notif.Method, key)
+			return true
+		}
+		sub.markAcked(agreed)
+		debug.Log("mcp-subs", "server=%s subscription %s acknowledged", c.name, key)
+		return true
+	}
+
+	key, ok := extractSubscriptionID(notif.Params)
+	if !ok {
+		return false
+	}
+	c.subMu.Lock()
+	sub := c.subs[key]
+	c.subMu.Unlock()
+	if sub == nil || !sub.acked.Load() {
+		debug.Log("mcp-subs", "server=%s notification %s for unknown/unacked subscription %s dropped",
+			c.name, notif.Method, key)
+		return true
+	}
+	return false
+}
+
+// parseSubscriptionAckParams decodes the acknowledged notification envelope:
+// {"_meta": {"io.modelcontextprotocol/subscriptionId": <id>},
+//
+//	"notifications": <filter>}. A missing notifications field yields an empty
+//
+// agreed filter (be liberal: full agreement).
+func parseSubscriptionAckParams(params json.RawMessage) (string, SubscriptionFilter, bool) {
+	if len(params) == 0 {
+		return "", SubscriptionFilter{}, false
+	}
+	var envelope struct {
+		Meta          map[string]json.RawMessage `json:"_meta"`
+		Notifications SubscriptionFilter         `json:"notifications"`
+	}
+	if err := json.Unmarshal(params, &envelope); err != nil {
+		return "", SubscriptionFilter{}, false
+	}
+	raw, ok := envelope.Meta[MetaKeySubscriptionID]
+	if !ok {
+		return "", SubscriptionFilter{}, false
+	}
+	key, ok := normalizeIDJSON(raw)
+	if !ok {
+		return "", SubscriptionFilter{}, false
+	}
+	return key, envelope.Notifications, true
+}
+
+// extractSubscriptionID reports whether params carries the listen-stream
+// correlation key, returning its normalized form.
+func extractSubscriptionID(params json.RawMessage) (string, bool) {
+	if len(params) == 0 {
+		return "", false
+	}
+	var envelope struct {
+		Meta map[string]json.RawMessage `json:"_meta"`
+	}
+	if err := json.Unmarshal(params, &envelope); err != nil {
+		return "", false
+	}
+	raw, ok := envelope.Meta[MetaKeySubscriptionID]
+	if !ok {
+		return "", false
+	}
+	return normalizeIDJSON(raw)
+}
diff --git a/internal/mcp/subscriptions_test.go b/internal/mcp/subscriptions_test.go
new file mode 100644
index 000000000..62d5628e6
--- /dev/null
+++ b/internal/mcp/subscriptions_test.go
@@ -0,0 +1,505 @@
+package mcp
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+)
+
+// Tests for the MCP 2026-07-28 subscriptions/listen client implementation
+// (subscriptions.go). The fake server is the pipe-based NDJSON harness from
+// issue994_test.go, extended with per-scenario behavior for the listen
+// request: ack-then-graceful-close, legacy -32601 downgrade, and silent
+// (cancel path) servers. The client-side read pump mirrors the real stdio
+// read loop: readMessage + deliverResponse for responses, processNotification
+// for notifications (which is where routeSubscriptionNotification runs).
+
+// subFakeServer is a scripted NDJSON stdio server for subscription tests.
+type subFakeServer struct {
+	client *Client
+	// seen receives every request/notification method the client writes,
+	// in order (buffered; non-blocking).
+	seen chan string
+	// onListen handles a subscriptions/listen request. write marshals and
+	// frames one JSON-RPC message to the client.
+	onListen func(write func(obj interface{}), reqID json.RawMessage)
+
+	reqRead    *os.File
+	respW      *os.File
+	cancelPump context.CancelFunc
+	pumpDone   chan struct{}
+	serverWG   sync.WaitGroup
+	mu         sync.Mutex
+	closed     bool
+}
+
+func newSubFakeServer(t *testing.T, onListen func(write func(obj interface{}), reqID json.RawMessage)) *subFakeServer {
+	t.Helper()
+	reqRead, reqWrite, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("request pipe: %v", err)
+	}
+	respRead, respWrite, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("response pipe: %v", err)
+	}
+
+	s := &subFakeServer{
+		client:   NewClient("sub-fake", "/bin/cat", nil),
+		seen:     make(chan string, 32),
+		onListen: onListen,
+		reqRead:  reqRead,
+		respW:    respWrite,
+		pumpDone: make(chan struct{}),
+	}
+	s.client.transport = "stdio"
+	s.client.stdin = reqWrite
+	s.client.reader = bufio.NewReader(respRead)
+
+	// Server loop: consume client-written NDJSON lines and run the scenario.
+	s.serverWG.Add(1)
+	go func() {
+		defer s.serverWG.Done()
+		scanner := bufio.NewScanner(reqRead)
+		for scanner.Scan() {
+			line := scanner.Bytes()
+			if len(line) == 0 {
+				continue
+			}
+			var msg struct {
+				ID     json.RawMessage `json:"id"`
+				Method string          `json:"method"`
+			}
+			if err := json.Unmarshal(line, &msg); err != nil {
+				continue
+			}
+			select {
+			case s.seen <- msg.Method:
+			default: // don't block the client on a full channel
+			}
+			switch msg.Method {
+			case MethodSubscriptionsListen:
+				if s.onListen != nil {
+					write := func(obj interface{}) {
+						data, err := json.Marshal(obj)
+						if err != nil {
+							return
+						}
+						_, _ = respWrite.Write(append(data, '\n'))
+					}
+					s.onListen(write, msg.ID)
+				}
+			}
+		}
+	}()
+
+	// Client-side read pump: mirrors the production stdio read loop
+	// (responses → waiters, notifications → processNotification).
+	pumpCtx, cancel := context.WithCancel(context.Background())
+	s.cancelPump = cancel
+	go func() {
+		defer close(s.pumpDone)
+		for {
+			msg, err := s.client.readMessage(pumpCtx)
+			if err != nil {
+				return
+			}
+			switch m := msg.(type) {
+			case *Notification:
+				s.client.processNotification(m)
+			case *Response:
+				s.client.deliverResponse(m)
+			}
+		}
+	}()
+
+	t.Cleanup(s.Close)
+	return s
+}
+
+// Close tears the fake server down without deadlocking on pipe reads.
+func (s *subFakeServer) Close() {
+	s.mu.Lock()
+	if s.closed {
+		s.mu.Unlock()
+		return
+	}
+	s.closed = true
+	s.mu.Unlock()
+	s.cancelPump()
+	_ = s.respW.Close()
+	_ = s.reqRead.Close()
+	select {
+	case <-s.pumpDone:
+	case <-time.After(5 * time.Second):
+	}
+	s.serverWG.Wait()
+}
+
+func (s *subFakeServer) seenCount(method string) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.seen) // approximation is fine; callers drain instead
+}
+
+func (s *subFakeServer) writeFromTest(obj interface{}) error {
+	data, err := json.Marshal(obj)
+	if err != nil {
+		return err
+	}
+	_, err = s.respW.Write(append(data, '\n'))
+	return err
+}
+
+// ackParams builds the acknowledged-notification params for reqID.
+func ackParams(reqID json.RawMessage, agreed string) json.RawMessage {
+	return json.RawMessage(fmt.Sprintf(
+		`{"_meta":{"%s":%s},"notifications":%s}`,
+		MetaKeySubscriptionID, reqID, agreed))
+}
+
+// TestSubscriptionListenStdioAckThenGraceful is the happy path: the server
+// acknowledges the requested filter subset, delivers nothing else, then
+// closes the stream gracefully. ListenSubscriptions must return only after
+// the ack, the Done channel must close on the terminal response, Err must be
+// nil, and the correlation registry must be empty afterwards.
+func TestSubscriptionListenStdioAckThenGraceful(t *testing.T) {
+	var reqID json.RawMessage
+	s := newSubFakeServer(t, func(write func(interface{}), id json.RawMessage) {
+		reqID = json.RawMessage(append([]byte(nil), id...))
+		write(map[string]interface{}{
+			"jsonrpc": "2.0",
+			"method":  NotificationSubscriptionsAcknowledged,
+			"params":  json.RawMessage(ackParams(id, `{"toolsListChanged":true}`)),
+		})
+		write(map[string]interface{}{
+			"jsonrpc": "2.0",
+			"id":      id,
+			"result":  map[string]interface{}{"resultType": "complete"},
+		})
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	filter := SubscriptionFilter{ToolsListChanged: true, ResourcesListChanged: true}
+	sub, err := s.client.ListenSubscriptions(ctx, filter)
+	if err != nil {
+		t.Fatalf("ListenSubscriptions: %v", err)
+	}
+	if !sub.acked.Load() {
+		t.Error("subscription should be acked on return")
+	}
+	if !sub.Agreed().ToolsListChanged {
+		t.Error("agreed filter should carry toolsListChanged")
+	}
+
+	select {
+	case <-sub.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("subscription did not end after graceful server closure")
+	}
+	if err := sub.Err(); err != nil {
+		t.Errorf("graceful closure should yield nil Err, got %v", err)
+	}
+	// Registry cleanup runs asynchronously (deliverResponse's spawned
+	// goroutine), so poll instead of asserting immediately.
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		s.client.subMu.Lock()
+		n := len(s.client.subs)
+		s.client.subMu.Unlock()
+		if n == 0 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("registry should drain after stream end, still %d", n)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if reqID == nil {
+		t.Error("server never received the listen request")
+	}
+}
+
+// TestSubscriptionLegacyServerDowngrade pins the -32601 contract: the error
+// surfaces to the caller, EnableModernSubscriptions reports false, and —
+// critically — the unsupported state latches so a legacy server never
+// receives a second subscriptions/listen.
+func TestSubscriptionLegacyServerDowngrade(t *testing.T) {
+	listens := 0
+	var mu sync.Mutex
+	s := newSubFakeServer(t, func(write func(interface{}), id json.RawMessage) {
+		mu.Lock()
+		listens++
+		mu.Unlock()
+		write(map[string]interface{}{
+			"jsonrpc": "2.0",
+			"id":      id,
+			"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+		})
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if s.client.EnableModernSubscriptions(ctx) {
+		t.Fatal("legacy server must not be reported as supported")
+	}
+	mu.Lock()
+	if listens != 1 {
+		mu.Unlock()
+		t.Fatalf("first attempt should hit the wire exactly once, saw %d", listens)
+	}
+	mu.Unlock()
+
+	// Latched: no second wire attempt for the auto path.
+	if s.client.EnableModernSubscriptions(ctx) {
+		t.Fatal("downgrade must latch")
+	}
+	mu.Lock()
+	got := listens
+	mu.Unlock()
+	if got != 1 {
+		t.Errorf("latched downgrade should not re-call the server, saw %d", got)
+	}
+	if s.client.subListenState.Load() != subStateUnsupported {
+		t.Error("subListenState should be unsupported after -32601")
+	}
+
+	// A direct ListenSubscriptions still surfaces the raw -32601 error.
+	var rpcErr *Error
+	_, err := s.client.ListenSubscriptions(ctx, SubscriptionFilter{ToolsListChanged: true})
+	if !errors.As(err, &rpcErr) || rpcErr.Code != -32601 {
+		t.Errorf("expected -32601 error surface, got %v", err)
+	}
+}
+
+// TestSubscriptionCancelSendsCancelledNotification covers the client-side
+// MUST: cancelling (here via the ack-wait context deadline) sends
+// notifications/cancelled with the listen request id and fully unwinds the
+// subscription (registry cleanup, Done closed).
+func TestSubscriptionCancelSendsCancelledNotification(t *testing.T) {
+	s := newSubFakeServer(t, nil) // silent server: never acks
+
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	_, err := s.client.ListenSubscriptions(ctx, SubscriptionFilter{ToolsListChanged: true})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("expected DeadlineExceeded, got %v", err)
+	}
+
+	// The cancelled notification must reach the wire.
+	deadline := time.After(5 * time.Second)
+	for {
+		select {
+		case m := <-s.seen:
+			if m == "notifications/cancelled" {
+				s.client.subMu.Lock()
+				n := len(s.client.subs)
+				s.client.subMu.Unlock()
+				if n != 0 {
+					t.Errorf("registry should be empty after cancel, got %d", n)
+				}
+				return
+			}
+		case <-deadline:
+			t.Fatal("notifications/cancelled never reached the server")
+		}
+	}
+}
+
+// TestRouteSubscriptionNotificationGating pins the MUST-gating matrix:
+//   - ack control messages are consumed and mark the subscription acked;
+//   - unknown/unacked subscription IDs are dropped (consumed);
+//   - valid correlated traffic falls through to legacy dispatch;
+//   - legacy traffic without _meta falls through untouched.
+func TestRouteSubscriptionNotificationGating(t *testing.T) {
+	c := NewClient("gating", "/bin/cat", nil)
+	cancelCtx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	id := NewIntID(42)
+	sub := &Subscription{
+		client:     c,
+		id:         &id,
+		ackCh:      make(chan struct{}),
+		done:       make(chan struct{}),
+		cancelCtx:  cancelCtx,
+		cancelFunc: cancelFunc,
+		requested:  SubscriptionFilter{ToolsListChanged: true},
+	}
+	c.addSubscription(subscriptionIDKey(sub.id), sub)
+
+	// 1. Ack is consumed and acks the subscription.
+	if !c.routeSubscriptionNotification(&Notification{
+		Method: NotificationSubscriptionsAcknowledged,
+		Params: ackParams(json.RawMessage(`42`), `{}`),
+	}) {
+		t.Error("ack must be consumed")
+	}
+	select {
+	case <-sub.ackCh:
+	default:
+		t.Error("ack channel should be closed after ack")
+	}
+	select {
+	case <-sub.done:
+		t.Error("ack must not end the subscription")
+	default:
+	}
+
+	// 2. Unknown subscription id → dropped.
+	if !c.routeSubscriptionNotification(&Notification{
+		Method: "notifications/tools/list_changed",
+		Params: json.RawMessage(`{"_meta":{"` + MetaKeySubscriptionID + `":999}}`),
+	}) {
+		t.Error("unknown subscription id must be consumed (dropped)")
+	}
+
+	// 3. Valid correlated traffic → NOT consumed, forwarded to legacy path.
+	if c.routeSubscriptionNotification(&Notification{
+		Method: "notifications/tools/list_changed",
+		Params: json.RawMessage(`{"_meta":{"` + MetaKeySubscriptionID + `":42}}`),
+	}) {
+		t.Error("valid correlated traffic must fall through to legacy dispatch")
+	}
+
+	// 4. Legacy traffic without _meta → untouched.
+	if c.routeSubscriptionNotification(&Notification{
+		Method: "notifications/message",
+		Params: json.RawMessage(`{"level":"info","data":"hi"}`),
+	}) {
+		t.Error("legacy traffic must fall through")
+	}
+
+	// 5. String-form subscription ids normalize the same way.
+	idStr := NewStringID("abc")
+	subStr := &Subscription{
+		client:     c,
+		id:         &idStr,
+		ackCh:      make(chan struct{}),
+		done:       make(chan struct{}),
+		cancelCtx:  cancelCtx,
+		cancelFunc: cancelFunc,
+	}
+	c.addSubscription(subscriptionIDKey(subStr.id), subStr)
+	// Unacked subscription ⇒ MUST drop; ack it first, then expect fall-through.
+	if !c.routeSubscriptionNotification(&Notification{
+		Method: NotificationSubscriptionsAcknowledged,
+		Params: ackParams(json.RawMessage(`"abc"`), `{}`),
+	}) {
+		t.Error("ack for string id must be consumed")
+	}
+	if c.routeSubscriptionNotification(&Notification{
+		Method: "notifications/resources/updated",
+		Params: json.RawMessage(`{"_meta":{"` + MetaKeySubscriptionID + `": "abc"},"uri":"file:///x"}`),
+	}) {
+		t.Error("string-id correlated traffic must fall through")
+	}
+
+	// 6. Malformed ack params are consumed, not forwarded.
+	if !c.routeSubscriptionNotification(&Notification{
+		Method: NotificationSubscriptionsAcknowledged,
+		Params: json.RawMessage(`{"no":"meta"}`),
+	}) {
+		t.Error("malformed ack must be consumed")
+	}
+}
+
+// TestSubscriptionFilterMissingFrom checks the ack-diff reporting logic.
+func TestSubscriptionFilterMissingFrom(t *testing.T) {
+	req := SubscriptionFilter{
+		ToolsListChanged:      true,
+		PromptsListChanged:    true,
+		ResourceSubscriptions: []string{"file:///a", "file:///b"},
+	}
+	agreed := SubscriptionFilter{
+		ToolsListChanged:      true,
+		ResourceSubscriptions: []string{"file:///a"},
+	}
+	got := req.missingFrom(agreed)
+	want := []string{"promptsListChanged", "resourceSubscriptions:file:///b"}
+	if len(got) != len(want) {
+		t.Fatalf("missingFrom = %v, want %v", got, want)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Errorf("missingFrom[%d] = %q, want %q", i, got[i], want[i])
+		}
+	}
+	if (SubscriptionFilter{}).missingFrom(agreed) != nil {
+		t.Error("empty request should grant nothing missing")
+	}
+}
+
+// TestSubscriptionEmptyFilterRejected guards the degenerate listen call.
+func TestSubscriptionEmptyFilterRejected(t *testing.T) {
+	c := NewClient("empty-filter", "/bin/cat", nil)
+	if _, err := c.ListenSubscriptions(context.Background(), SubscriptionFilter{}); err == nil {
+		t.Fatal("empty filter must be rejected client-side")
+	}
+}
+
+// TestSubscriptionWSUnsupported documents the transport limitation.
+func TestSubscriptionWSUnsupported(t *testing.T) {
+	c := NewClient("ws-sub", "/bin/cat", nil)
+	c.transport = "ws"
+	_, err := c.ListenSubscriptions(context.Background(), SubscriptionFilter{ToolsListChanged: true})
+	if err == nil || !isUnsupportedTransportErr(err) {
+		t.Fatalf("ws transport must be rejected, got %v", err)
+	}
+}
+
+func isUnsupportedTransportErr(err error) bool {
+	type transportErr interface{ error }
+	_ = transportErr(nil)
+	return err != nil && containsStr(err.Error(), "unsupported on transport")
+}
+
+func containsStr(s, sub string) bool {
+	return len(s) >= len(sub) && (func() bool {
+		for i := 0; i+len(sub) <= len(s); i++ {
+			if s[i:i+len(sub)] == sub {
+				return true
+			}
+		}
+		return false
+	})()
+}
+
+// TestSubscriptionAbortClosesAll pins the teardown contract: Abort ends every
+// open subscription with a terminal error so Wait-style consumers unblock
+// even though the server never closed the stream.
+func TestSubscriptionAbortClosesAll(t *testing.T) {
+	var ackSend func(write func(interface{}), id json.RawMessage)
+	s := newSubFakeServer(t, nil) // handler wired below (needs s)
+	ackSend = func(write func(interface{}), id json.RawMessage) {
+		// Ack but never close — the stream stays open until Abort.
+		_ = s.writeFromTest(map[string]interface{}{
+			"jsonrpc": "2.0",
+			"method":  NotificationSubscriptionsAcknowledged,
+			"params":  json.RawMessage(ackParams(id, `{}`)),
+		})
+	}
+	s.onListen = ackSend
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	sub, err := s.client.ListenSubscriptions(ctx, SubscriptionFilter{ToolsListChanged: true})
+	if err != nil {
+		t.Fatalf("ListenSubscriptions: %v", err)
+	}
+	s.client.Abort()
+	select {
+	case <-sub.Done():
+		if sub.Err() == nil {
+			t.Error("abort teardown should carry a terminal error")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("subscription did not end after Abort")
+	}
+}
diff --git a/internal/plugin/mcp_loader.go b/internal/plugin/mcp_loader.go
index 68adea761..90d262330 100644
--- a/internal/plugin/mcp_loader.go
+++ b/internal/plugin/mcp_loader.go
@@ -295,11 +295,32 @@ func (m *MCPPlugin) Connect(ctx context.Context) (*mcp.Adapter, error) {
 	m.prompts = prompts
 	m.resources = resources
 	m.setupNotificationHandler(client)
+	// MCP 2026-07-28: self-detecting upgrade to the correlated subscription
+	// stream (subscriptions/listen). Legacy servers answer -32601 and are
+	// downgraded permanently for this client; modern servers ack the filter
+	// and subsequent list-change notifications arrive correlated (see
+	// internal/mcp/subscriptions.go). The legacy dispatch above is unchanged
+	// either way.
+	m.setupModernSubscriptions(client)
 	m.startReconnectWatcher(client)
 	m.startWSHealthProbe(client)
 	return m.adapter, nil
 }
 
+// setupModernSubscriptions attempts the MCP 2026-07-28 subscriptions/listen
+// upgrade for this server connection. It is deliberately self-detecting and
+// best-effort: success opens a correlated notification stream, a legacy
+// server's method-not-found response downgrades silently, and any other
+// failure only costs a debug log. Called from the shared connect path so
+// auto-reconnect re-opens the stream automatically.
+func (m *MCPPlugin) setupModernSubscriptions(client *mcp.Client) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if client.EnableModernSubscriptions(ctx) {
+		debug.Log("mcp-notif", "server=%s modern subscription stream active", m.cfg.Name)
+	}
+}
+
 // setupNotificationHandler registers a notification handler on the MCP client
 // to process server-initiated notifications:
 //   - notifications/tools/list_changed: triggers hot tool list refresh
diff --git a/internal/plugin/mcp_loader_test.go b/internal/plugin/mcp_loader_test.go
index 2e74ad74e..4375c131f 100644
--- a/internal/plugin/mcp_loader_test.go
+++ b/internal/plugin/mcp_loader_test.go
@@ -30,7 +30,18 @@ func TestMCPManagerConnectAllTimesOutHungServer(t *testing.T) {
 		if req.Method == "initialize" {
 			time.Sleep(100 * time.Millisecond)
 			w.Header().Set("Content-Type", "application/json")
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}}`))
+			writeEcho(w, req.ID, `{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}`)
+			return
+		}
+		if req.Method == mcp.MethodSubscriptionsListen {
+			// MCP 2026-07-28 probe: legacy mock — spec downgrade error.
+			w.Header().Set("Content-Type", "application/json")
+			echo, _ := json.Marshal(map[string]interface{}{
+				"jsonrpc": "2.0",
+				"id":      req.ID,
+				"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+			})
+			_, _ = w.Write(echo)
 			return
 		}
 		t.Fatalf("unexpected method %s", req.Method)
@@ -112,15 +123,24 @@ func TestMCPPluginInfoIncludesPromptAndResourceNames(t *testing.T) {
 		w.Header().Set("Content-Type", "application/json")
 		switch req.Method {
 		case "initialize":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}}`))
+			writeEcho(w, req.ID, `{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}`)
 		case "notifications/initialized":
 			w.WriteHeader(http.StatusNoContent)
 		case "tools/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"fetch","description":"Fetch","inputSchema":{"type":"object"}}]}}`))
+			writeEcho(w, req.ID, `{"tools":[{"name":"fetch","description":"Fetch","inputSchema":{"type":"object"}}]}`)
 		case "prompts/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":3,"result":{"prompts":[{"name":"summarize"},{"name":"translate"}]}}`))
+			writeEcho(w, req.ID, `{"prompts":[{"name":"summarize"},{"name":"translate"}]}`)
 		case "resources/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":4,"result":{"resources":[{"name":"docs"},{"uri":"file:///tmp/readme.md"}]}}`))
+			writeEcho(w, req.ID, `{"resources":[{"name":"docs"},{"uri":"file:///tmp/readme.md"}]}`)
+		case mcp.MethodSubscriptionsListen:
+			// MCP 2026-07-28 probe: this mock is legacy — downgrade the
+			// client with the spec method-not-found error.
+			echo, _ := json.Marshal(map[string]interface{}{
+				"jsonrpc": "2.0",
+				"id":      req.ID,
+				"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+			})
+			_, _ = w.Write(echo)
 		default:
 			t.Fatalf("unexpected method %s", req.Method)
 		}
@@ -163,15 +183,24 @@ func TestMCPPluginInfoDoesNotBlockWhileConnectIsInFlight(t *testing.T) {
 		case "initialize":
 			close(initialized)
 			<-release
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}}`))
+			writeEcho(w, req.ID, `{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}`)
 		case "notifications/initialized":
 			w.WriteHeader(http.StatusNoContent)
 		case "tools/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":2,"result":{"tools":[]}}`))
+			writeEcho(w, req.ID, `{"tools":[]}`)
 		case "prompts/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":3,"result":{"prompts":[]}}`))
+			writeEcho(w, req.ID, `{"prompts":[]}`)
 		case "resources/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":4,"result":{"resources":[]}}`))
+			writeEcho(w, req.ID, `{"resources":[]}`)
+		case mcp.MethodSubscriptionsListen:
+			// MCP 2026-07-28 probe: this mock is legacy — downgrade the
+			// client with the spec method-not-found error.
+			echo, _ := json.Marshal(map[string]interface{}{
+				"jsonrpc": "2.0",
+				"id":      req.ID,
+				"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+			})
+			_, _ = w.Write(echo)
 		default:
 			t.Fatalf("unexpected method %s", req.Method)
 		}
@@ -227,19 +256,28 @@ func TestMCPManagerPromptAndResourceAccess(t *testing.T) {
 		w.Header().Set("Content-Type", "application/json")
 		switch req.Method {
 		case "initialize":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}}`))
+			writeEcho(w, req.ID, `{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}`)
 		case "notifications/initialized":
 			w.WriteHeader(http.StatusNoContent)
 		case "tools/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":2,"result":{"tools":[]}}`))
+			writeEcho(w, req.ID, `{"tools":[]}`)
 		case "prompts/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":3,"result":{"prompts":[{"name":"summarize"}]}}`))
+			writeEcho(w, req.ID, `{"prompts":[{"name":"summarize"}]}`)
 		case "resources/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":4,"result":{"resources":[{"uri":"docs"}]}}`))
+			writeEcho(w, req.ID, `{"resources":[{"uri":"docs"}]}`)
 		case "prompts/get":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":5,"result":{"description":"Prompt","messages":[{"role":"user","content":{"type":"text","text":"hello prompt"}}]}}`))
+			writeEcho(w, req.ID, `{"description":"Prompt","messages":[{"role":"user","content":{"type":"text","text":"hello prompt"}}]}`)
 		case "resources/read":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":6,"result":{"contents":[{"uri":"docs","mimeType":"text/plain","text":"hello resource"}]}}`))
+			writeEcho(w, req.ID, `{"contents":[{"uri":"docs","mimeType":"text/plain","text":"hello resource"}]}`)
+		case mcp.MethodSubscriptionsListen:
+			// MCP 2026-07-28 probe: this mock is legacy — downgrade the
+			// client with the spec method-not-found error.
+			echo, _ := json.Marshal(map[string]interface{}{
+				"jsonrpc": "2.0",
+				"id":      req.ID,
+				"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+			})
+			_, _ = w.Write(echo)
 		default:
 			t.Fatalf("unexpected method %s", req.Method)
 		}
@@ -285,15 +323,24 @@ func TestMCPManagerInstallAddsServer(t *testing.T) {
 		w.Header().Set("Content-Type", "application/json")
 		switch req.Method {
 		case "initialize":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}}`))
+			writeEcho(w, req.ID, `{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}`)
 		case "notifications/initialized":
 			w.WriteHeader(http.StatusNoContent)
 		case "tools/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"fetch","description":"Fetch","inputSchema":{"type":"object"}}]}}`))
+			writeEcho(w, req.ID, `{"tools":[{"name":"fetch","description":"Fetch","inputSchema":{"type":"object"}}]}`)
 		case "prompts/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":3,"result":{"prompts":[]}}`))
+			writeEcho(w, req.ID, `{"prompts":[]}`)
 		case "resources/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":4,"result":{"resources":[]}}`))
+			writeEcho(w, req.ID, `{"resources":[]}`)
+		case mcp.MethodSubscriptionsListen:
+			// MCP 2026-07-28 probe: this mock is legacy — downgrade the
+			// client with the spec method-not-found error.
+			echo, _ := json.Marshal(map[string]interface{}{
+				"jsonrpc": "2.0",
+				"id":      req.ID,
+				"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+			})
+			_, _ = w.Write(echo)
 		default:
 			t.Fatalf("unexpected method %s", req.Method)
 		}
@@ -331,15 +378,24 @@ func TestMCPManagerUninstallRemovesServerAndTools(t *testing.T) {
 		w.Header().Set("Content-Type", "application/json")
 		switch req.Method {
 		case "initialize":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}}`))
+			writeEcho(w, req.ID, `{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mock","version":"1.0.0"}}`)
 		case "notifications/initialized":
 			w.WriteHeader(http.StatusNoContent)
 		case "tools/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"fetch","description":"Fetch","inputSchema":{"type":"object"}}]}}`))
+			writeEcho(w, req.ID, `{"tools":[{"name":"fetch","description":"Fetch","inputSchema":{"type":"object"}}]}`)
 		case "prompts/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":3,"result":{"prompts":[]}}`))
+			writeEcho(w, req.ID, `{"prompts":[]}`)
 		case "resources/list":
-			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":4,"result":{"resources":[]}}`))
+			writeEcho(w, req.ID, `{"resources":[]}`)
+		case mcp.MethodSubscriptionsListen:
+			// MCP 2026-07-28 probe: this mock is legacy — downgrade the
+			// client with the spec method-not-found error.
+			echo, _ := json.Marshal(map[string]interface{}{
+				"jsonrpc": "2.0",
+				"id":      req.ID,
+				"error":   map[string]interface{}{"code": -32601, "message": "Method not found"},
+			})
+			_, _ = w.Write(echo)
 		default:
 			t.Fatalf("unexpected method %s", req.Method)
 		}
@@ -369,3 +425,23 @@ func TestMCPManagerUninstallRemovesServerAndTools(t *testing.T) {
 		t.Fatal("expected MCP tool to be unregistered after uninstall")
 	}
 }
+
+// writeEcho frames a JSON-RPC success payload echoing the request id. The
+// client assigns request ids dynamically (and since the subscriptions/listen
+// probe occupies one slot, hardcoded ids in mocks no longer line up), so
+// every mock response must echo the actual request id instead.
+func writeEcho(w http.ResponseWriter, id *mcp.ID, result string) {
+	resp, err := json.Marshal(struct {
+		JSONRPC string          `json:"jsonrpc"`
+		ID      *mcp.ID         `json:"id"`
+		Result  json.RawMessage `json:"result"`
+	}{
+		JSONRPC: "2.0",
+		ID:      id,
+		Result:  json.RawMessage(result),
+	})
+	if err != nil {
+		return
+	}
+	_, _ = w.Write(resp)
+}