feat(19.7-02): Redis sessions, event fan-out, stateless protocol path and resource metadata

- store_redis.go: records with PX TTLs, per-binding sets, pub/sub events, opened lazily from redis.url under cooler's prefix (nothing dials at mount) - events.go: pact.MCPSubjectChanged republished through the store; every instance recomputes the subject's sessions and stateless servers (list_changed) or ends them on revocation; the link cache is dropped - stateless.go: the SDK's stateless handler over per-identity servers for 2026-07-28 clients, routed by the protocol header or server/discover - prm.go: RFC 9728 metadata at both well-known paths; long GET streams re-verify their bearer every minute - TestPierSessionsSmoke, TestPierFanOutSmoke, TestPierStatelessSmoke on miniredis; README features, configuration and operational notes
2026-10-22 18:23:43 +02:00
parent cfdbaf0e47
commit 806e87c510
11 changed files with 1133 additions and 32 deletions
--- a/modules/pier/README.md
+++ b/modules/pier/README.md
@@ -12,7 +12,7 @@ pier mounts a streamable HTTP MCP endpoint (POST, GET and DELETE at `mcp.path`,

 Every request to the endpoint is authorized afresh from its bearer: an OAuth access token through the `pact.MCPTokenVerifier` the OAuth server plugin publishes, or, behind `mcp.allow_personal_tokens`, a personal API token (an admin `g15_` token through cabana, a frontend token through the published `pact.MCPPersonalTokenResolver`). The resulting `pier.Identity` names a frontend subject and, when the grant carries the `backend` scope and the published `pact.MCPLinkedBackendResolver` confirms a link, a backend subject. `tools/list` is computed for that identity: a tool needs a subject under its guard, a grant without `write` sees only read-only tools, and the guard provider's own check (the same logic its route middleware enforces) decides the rest. A `tools/call` runs the bound route's compiled middleware chain in-process with the `bouncer.Dispatch` marker carrying the principal the guard provider resolved; the inner request never carries an `Authorization` or `Cookie` header. A 2xx answer becomes text plus structured content, anything else a tool error carrying the route's own body.

-Sessions (`Mcp-Session-Id`) are bound to the identity that opened them (client and user, or the personal token), live in a store with a sliding idle TTL, and are served by a per-session SDK server on whichever instance receives the request, rehydrated from the stored session state when it is not local. The memory store serves one instance; the Redis store (next task) shares sessions and events across instances.
+Sessions (`Mcp-Session-Id`) are bound to the identity that opened them (client and user, or the personal token), live in a store with a sliding idle TTL, and are served by a per-session SDK server on whichever instance receives the request, rehydrated from the stored session state when it is not local. The memory store serves one instance; the Redis store shares sessions across instances and fans subject events out to every instance, so a permission or link change pushes `notifications/tools/list_changed` to the sessions it concerns and a revocation ends them everywhere. Clients of protocol 2026-07-28 and later, which have no sessions, are served statelessly from the same path by the SDK's stateless handler over per-identity servers, sharing identity, dispatch and filtering with the stateful path.

 ## Role in the framework

@@ -35,7 +35,11 @@ Use cases:
 - Built-in `backend` guard provider (`pier.BackendGuard`): routes behind cabana's `backend` middleware, listing through `cabana.Allows` on the administrator's principal from `cabana.BackendPrincipal`, an admin personal token's scope limit computed per dispatched route method with `cabana.ScopeLimitFor`; a plugin that registers its own `backend` provider wins.
 - Identity per request (`pier.Identity`): `oas_at_` tokens through the published verifier, personal tokens only with `mcp.allow_personal_tokens`, the backend subject through the published link resolver with an in-process cache of at most `pier.LinkCacheTTL`; the binding key `oauth:<client_id>:<user_id>`, `pat:user:<token_id>` or `pat:backend:<token_id>` survives access-token rotation.
 - Per-identity SDK server plus a receiving middleware that re-filters `tools/list` and `prompts/list` and refuses `tools/call` and `prompts/get` for the identity of the current request, so a permission change takes effect on the next request (D-08, D-25).
- Stateful sessions: `initialize` opens a session (429 above `mcp.sessions.max_per_binding`), a request with an unknown id answers 404 (the client re-initialises), another identity presenting the id answers 403 `session_binding`, DELETE ends the session everywhere, the record's TTL slides on every request, a session initialised on one instance is rehydrated on another from the stored `ServerSessionState`.
+- Stateful sessions (D-25, D-27): `initialize` opens a session (429 above `mcp.sessions.max_per_binding`), a request with an unknown id answers 404 (the client re-initialises), another identity presenting the id answers 403 `session_binding`, DELETE ends the session everywhere, the record's TTL slides on every request, a session initialised on one instance is rehydrated on another from the stored `ServerSessionState`. A standalone GET stream re-verifies its bearer every minute and ends when it no longer resolves to the session's binding.
+- Session stores: `memory` (one instance; records in the process) or `redis` (records `<prefix>mcp:s:<id>` with `PX` TTLs, per-binding sets `<prefix>mcp:b:<binding>`, events on channel `<prefix>mcp:events`, under `cooler.KeyPrefix`); the Redis client is opened on the first request from `redis.url`, never at mount, so `route:list` and tests never dial, and a missing URL is a 503 logged once.
+- Fan-out (D-26): every `pact.MCPSubjectChanged` fired on the application is published through the store and applied on every instance: a permission or link change recomputes the tools and prompts of the subject's sessions and stateless servers (the SDK then sends `notifications/tools/list_changed` and `notifications/prompts/list_changed`), a revocation or deletion closes those sessions, drops their records and evicts the stateless servers; a link event also drops the link cache at once.
+- Hybrid protocol (D-32): a request whose `Mcp-Protocol-Version` is 2026-07-28 or later, or whose first JSON-RPC method is `server/discover`, goes to the SDK's stateless handler over an in-process LRU of per-identity servers keyed by binding (capacity `mcp.sessions.max_per_binding` times 16); every other request goes to the stateful router. Both run behind the same bearer verification.
+- Protected resource metadata (RFC 9728, `pier.PRMScopes`): `GET /.well-known/oauth-protected-resource` and `GET /.well-known/oauth-protected-resource<mcp.path>` answer the resource, the authorization server (the published verifier's resource origin, else `app.url`), the supported scopes and the bearer method; a `mcp.resource` that differs from the verifier's resource is a boot warning.
 - Prompts (D-28): prefixed and filtered like tools, required arguments checked before `Render` runs; a prompt never dispatches a route.
 - A 401 carries `WWW-Authenticate: Bearer resource_metadata="<app.url>/.well-known/oauth-protected-resource<mcp.path>", scope="read"`; a browser request whose `Origin` is not in `mcp.allowed_origins` is refused with 403 before any body is read.
 - One audit line per tool call (`pier: tool call`: tool, guard, user id, admin id, client id, token kind, session prefix, status, duration) and one per session step (`pier: session`), never arguments, bodies or tokens.
@@ -103,6 +107,7 @@ An MCP client then connects to `https://example.com/mcp` with an OAuth access to
 | `pier.Record` | One stored session: binding, session state JSON and creation time. |
 | `pier.Event` | One instance-to-instance event: a kind (`pier.EventEnded` or a `pact.MCPSubjectChanged` kind), guard, user id, session id and publishing instance. |
 | `pier.LogComponent` | `pier`, the diary component of the audit lines. |
+| `pier.PRMScopes` | The `scopes_supported` list of the protected resource metadata (`read`, `write`, `ai`, `offline_access`, `backend`). |

 Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`, `pier.AllowPersonalTokensKey`, `pier.SessionStoreKey`, `pier.SessionIdleTTLKey`, `pier.SessionMaxPerBindingKey`, `pier.ResourceKey`, `pier.ThrottleKey`, `pier.AllowedOriginsKey`, `pier.CallTimeoutKey`, `pier.MaxResultBytesKey`, with defaults `pier.DefaultPath`, `pier.DefaultSessionStore`, `pier.DefaultIdleTTL`, `pier.DefaultMaxPerBinding`, `pier.DefaultThrottle`, `pier.DefaultCallTimeout` and `pier.DefaultMaxResultBytes`, and store names `pier.StoreMemory` and `pier.StoreRedis`.

@@ -113,7 +118,7 @@ Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`,
 | `mcp.enabled` | `false` | Mounts the endpoint. `true` in a binary built without the `summercms_mcp` tag fails boot naming the tag. |
 | `mcp.path` | `/mcp` | The endpoint path: an absolute URL path without scheme, host, query, wildcard or dot segments. A route a plugin registered there fails boot naming the plugin. |
 | `mcp.allow_personal_tokens` | `false` | Lets personal API tokens open sessions: an admin `g15_` token yields backend tools only, a frontend token (through the published resolver) frontend tools only, each capped by the token's own scope. OAuth tokens are always accepted. |
-| `mcp.sessions.store` | `memory` | `memory` (one instance; sessions end with the process) or `redis` (shared through `redis.url`; the Redis store arrives with the next task). |
+| `mcp.sessions.store` | `memory` | `memory` (one instance; sessions end with the process and events reach only its own sessions) or `redis` (sessions and events shared by every instance through `redis.url`, opened on the first request). |
 | `mcp.sessions.idle_ttl` | `30m` | Sliding idle TTL of a session record. |
 | `mcp.sessions.max_per_binding` | `20` | Open sessions one binding (client and user, or personal token) may hold; the next `initialize` answers 429. |
 | `mcp.resource` | `app.url` + `mcp.path` | The resource indicator the endpoint identifies itself by; must equal the OAuth server's MCP resource. Required (or `app.url`) when enabled. |
@@ -124,6 +129,8 @@ Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`,

 Environment: `SUMMER_MCP__ENABLED`, `SUMMER_MCP__PATH`, `SUMMER_MCP__SESSIONS__STORE` and so on.

+Operational notes: a single instance (the demo deployment) runs on the memory store; several instances behind one load balancer need `mcp.sessions.store: redis` so a session opened on one instance is served by another and a permission change reaches every session. A reverse proxy must route `mcp.path` and both `/.well-known/oauth-protected-resource` paths to the binary and must not buffer the endpoint's responses (server-sent events). Every tool call leaves one `pier: tool call` line and every session step one `pier: session` line under the `pier` component.
+
 ## Dependencies

 - `github.com/modelcontextprotocol/go-sdk` v1.8.0 (`mcp`, `auth`, `oauthex`, `jsonrpc`): the server, sessions, streamable HTTP transport, bearer middleware and metadata types; decided in `.planning/notes/2026-10-22-mcp-go-sdk.md`.
@@ -134,4 +141,4 @@ Only this module (and surf's tagged `pier_on.go`) imports the SDK; `pact` stays

 ## Testing

-`go test -short ./modules/pier/...` runs without Docker, Redis or a build tag: `TestPierTracerSmoke` builds a real surf router with a fake plugin, mounts pier through an in-test `pier.Host` adapter, connects the SDK client over `httptest` and proves `tools/list`, `tools/call` through the compiled guard chain with the dispatch marker (no `Authorization` or `Cookie` reaches the route), the 401 challenge, and the boot errors for a duplicate name and an unknown route. `go test -short -tags summercms_mcp ./modules/pier/... ./modules/surf/...` runs the same tests with the SDK linked into surf.
+`go test -short ./modules/pier/...` runs without Docker, a real Redis or a build tag: `TestPierTracerSmoke` builds a real surf router with a fake plugin, mounts pier through an in-test `pier.Host` adapter, connects the SDK client over `httptest` and proves `tools/list`, `tools/call` through the compiled guard chain with the dispatch marker (no `Authorization` or `Cookie` reaches the route), the 401 challenge, and the boot errors for a duplicate name and an unknown route; `TestPierSessionsSmoke` runs two instances on one miniredis (rehydration, the binding refusal, DELETE, the idle TTL, the per-binding cap); `TestPierFanOutSmoke` receives `list_changed` on one instance for an event fired on the other and sees a revocation end the session; `TestPierStatelessSmoke` connects the newest SDK client statelessly next to a 2025-11-25 client with a session. `go test -short -tags summercms_mcp ./modules/pier/... ./modules/surf/...` runs the same tests with the SDK linked into surf.
--- a/modules/pier/config.go
+++ b/modules/pier/config.go
@@ -49,12 +49,12 @@ const (

 // Defaults of the mcp.* keys.
 const (
-	DefaultPath          = "/mcp"
-	DefaultSessionStore  = "memory"
-	DefaultIdleTTL       = 30 * time.Minute
-	DefaultMaxPerBinding = 20
-	DefaultThrottle      = "120,1"
-	DefaultCallTimeout   = 30 * time.Second
+	DefaultPath           = "/mcp"
+	DefaultSessionStore   = "memory"
+	DefaultIdleTTL        = 30 * time.Minute
+	DefaultMaxPerBinding  = 20
+	DefaultThrottle       = "120,1"
+	DefaultCallTimeout    = 30 * time.Second
 	DefaultMaxResultBytes = int64(1 << 20)
 )

--- a/modules/pier/events.go
+++ b/modules/pier/events.go
@@ -0,0 +1,184 @@
+package pier
+
+import (
+	"context"
+
+	"git.golem15.com/golem15/summercms/modules/pact"
+)
+
+// listenEvents republishes every pact.MCPSubjectChanged fired on the
+// application through the store, so every instance (this one included)
+// applies it in handleEvent; when publishing fails the event is applied
+// locally at least.
+func (p *Pier) listenEvents() {
+	if p.app == nil || p.app.Events == nil {
+		return
+	}
+	p.app.Events.Listen(PluginID, func(ctx context.Context, ev *pact.MCPSubjectChanged) error {
+		if ev == nil || ev.Kind == "" {
+			return nil
+		}
+		event := Event{Kind: ev.Kind, Guard: ev.Guard, UserID: ev.UserID, Instance: p.instance}
+		if err := p.store.Publish(ctx, event); err != nil {
+			p.handleEvent(event)
+		}
+		return nil
+	})
+}
+
+// handleEvent applies one event on this instance (D-26, D-27): a session
+// ended elsewhere is closed here; a permission or link change recomputes
+// the tools and prompts of every local session and stateless server of
+// that subject, which makes the SDK send the list_changed notifications; a
+// revocation or deletion closes those sessions, drops their records and
+// evicts the stateless servers. A link event also drops the link cache.
+func (p *Pier) handleEvent(ev Event) {
+	ctx := context.Background()
+	switch ev.Kind {
+	case EventEnded:
+		if ev.SessionID != "" && ev.Instance != p.instance {
+			p.sessions.closeLocal(ev.SessionID)
+		}
+	case pact.MCPSubjectRevoked, pact.MCPSubjectDeleted:
+		for _, ls := range p.sessions.matching(ev.Guard, ev.UserID) {
+			_ = p.store.Delete(ctx, ls.id)
+			p.sessions.closeLocal(ls.id)
+			logSession(logger(), "revoked", ls.id, ls.binding)
+		}
+		if p.stateless != nil {
+			p.stateless.evict(ev.Guard, ev.UserID)
+		}
+	case pact.MCPSubjectLink, pact.MCPSubjectPermissions:
+		if ev.Kind == pact.MCPSubjectLink {
+			if ev.Guard == BackendGuard {
+				p.identities.forgetAllLinks()
+			} else {
+				p.identities.forgetLink(ev.UserID)
+			}
+		}
+		for _, ls := range p.sessions.matching(ev.Guard, ev.UserID) {
+			p.recomputeSession(ctx, ls, ev.Kind == pact.MCPSubjectLink)
+		}
+		if p.stateless != nil {
+			for _, e := range p.stateless.matching(ev.Guard, ev.UserID) {
+				p.recomputeStateless(ctx, e, ev.Kind == pact.MCPSubjectLink)
+			}
+		}
+	}
+}
+
+// recomputeSession re-filters a local session's identity (re-resolving
+// the backend link after a link event) and applies the difference with
+// AddTool, RemoveTools, AddPrompt and RemovePrompts.
+func (p *Pier) recomputeSession(ctx context.Context, ls *localSession, relink bool) {
+	ls.mu.Lock()
+	id := ls.identity
+	if relink {
+		p.identities.refreshBackend(ctx, &id)
+		ls.identity = id
+	}
+	tools, prompts := p.catalog.allowed(ctx, id)
+	applyDiff(ls.server, p, id, tools, prompts, ls.tools, ls.prompts)
+	ls.mu.Unlock()
+}
+
+// recomputeStateless does the same for one per-identity stateless server.
+func (p *Pier) recomputeStateless(ctx context.Context, e *statelessEntry, relink bool) {
+	e.mu.Lock()
+	id := e.identity
+	if relink {
+		p.identities.refreshBackend(ctx, &id)
+		e.identity = id
+	}
+	tools, prompts := p.catalog.allowed(ctx, id)
+	applyDiff(e.server, p, id, tools, prompts, e.tools, e.prompts)
+	e.mu.Unlock()
+}
+
+// applyDiff brings a server's tool and prompt sets to the allowed ones,
+// updating the tracking maps in place.
+func applyDiff(srv serverFeatures, p *Pier, id Identity, tools []*tool, prompts []*prompt, haveTools, havePrompts map[string]bool) {
+	wantTools := map[string]bool{}
+	for _, t := range tools {
+		wantTools[t.name] = true
+		if !haveTools[t.name] {
+			srv.AddTool(t.desc, p.toolHandler(t, id))
+			haveTools[t.name] = true
+		}
+	}
+	var dropTools []string
+	for name := range haveTools {
+		if !wantTools[name] {
+			dropTools = append(dropTools, name)
+			delete(haveTools, name)
+		}
+	}
+	if len(dropTools) > 0 {
+		srv.RemoveTools(dropTools...)
+	}
+	wantPrompts := map[string]bool{}
+	for _, pr := range prompts {
+		wantPrompts[pr.name] = true
+		if !havePrompts[pr.name] {
+			srv.AddPrompt(pr.desc, p.promptHandler(pr))
+			havePrompts[pr.name] = true
+		}
+	}
+	var dropPrompts []string
+	for name := range havePrompts {
+		if !wantPrompts[name] {
+			dropPrompts = append(dropPrompts, name)
+			delete(havePrompts, name)
+		}
+	}
+	if len(dropPrompts) > 0 {
+		srv.RemovePrompts(dropPrompts...)
+	}
+}
+
+// matching lists the local sessions whose identity includes the subject.
+func (sr *sessionRouter) matching(guard string, userID uint) []*localSession {
+	sr.mu.Lock()
+	defer sr.mu.Unlock()
+	var out []*localSession
+	for _, ls := range sr.local {
+		ls.mu.Lock()
+		hit := ls.identity.has(guard, userID)
+		ls.mu.Unlock()
+		if hit {
+			out = append(out, ls)
+		}
+	}
+	return out
+}
+
+// refreshBackend re-resolves an OAuth identity's backend subject through
+// the link resolver (after a link event), adding or dropping it.
+func (r *identityResolver) refreshBackend(ctx context.Context, id *Identity) {
+	if id.TokenKind != TokenKindOAuth || id.Frontend == nil {
+		return
+	}
+	id.Backend = nil
+	for _, s := range id.Scopes {
+		if s != "backend" {
+			continue
+		}
+		if link, ok := r.linkedBackend(ctx, id.Frontend.UserID); ok {
+			id.Backend = &pact.MCPSubject{
+				Guard:    BackendGuard,
+				UserID:   link.ID,
+				Scopes:   id.Scopes,
+				ClientID: id.Frontend.ClientID,
+				TokenID:  id.Frontend.TokenID,
+			}
+		}
+		return
+	}
+}
+
+// forgetAllLinks drops the whole link cache.
+func (r *identityResolver) forgetAllLinks() {
+	r.mu.Lock()
+	r.links = map[uint]linkEntry{}
+	r.mu.Unlock()
+}
--- a/modules/pier/pier.go
+++ b/modules/pier/pier.go
@@ -17,11 +17,18 @@ import (
 	"runtime/debug"
 	"strings"

+	"time"
+
 	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/cooler"
 	"git.golem15.com/golem15/summercms/modules/party"
 	"github.com/modelcontextprotocol/go-sdk/auth"
 )

+// streamRecheckInterval is how often a long-lived GET stream re-verifies
+// its bearer (Pitfall 10); a failing bearer closes the stream.
+const streamRecheckInterval = time.Minute
+
 // frameworkModule is the module path whose version names the server.
 const frameworkModule = "git.golem15.com/golem15/summercms"

@@ -43,7 +50,9 @@ type Pier struct {
 	dispatch   *dispatcher
 	store      Store
 	sessions   *sessionRouter
+	stateless  *statelessPath
 	handler    http.Handler
+	recheck    time.Duration
 	prmURL     string
 	appName    string
 	version    string
@@ -89,16 +98,18 @@ func Mount(host Host, app *backpack.App, plugins []party.Plugin) error {
 		app:      app,
 		host:     host,
 		catalog:  cat,
-		store:    newMemoryStore(),
 		appName:  appName(app),
 		version:  frameworkVersion(),
 		instance: rand.Text(),
+		recheck:  streamRecheckInterval,
 	}
 	p.identities = newIdentityResolver(app, cfg, frontend)
 	p.dispatch = &dispatcher{host: host, app: app, cfg: cfg}
 	p.sessions = newSessionRouter(p)
+	p.stateless = newStatelessPath(p)
+	p.store = p.newStore()
 	p.prmURL = p.metadataURL()
-	p.handler = p.originCheck(p.bodyLimit(p.bearer(p.sessions)))
+	p.handler = p.originCheck(p.bodyLimit(p.bearer(http.HandlerFunc(p.route))))

 	methods := []string{http.MethodPost, http.MethodGet, http.MethodDelete}
 	for _, m := range methods {
@@ -111,9 +122,46 @@ func Mount(host Host, app *backpack.App, plugins []party.Plugin) error {
 			return err
 		}
 	}
+	if err := p.mountPRM(); err != nil {
+		return err
+	}
+	p.listenEvents()
 	return nil
 }

+// newStore builds the session store mcp.sessions.store selects: the memory
+// store, subscribed to its own events at once, or a Redis store opened on
+// first use from redis.url under cooler's key prefix (nothing dials at
+// mount; a missing URL is a 503 on the first request, logged once).
+func (p *Pier) newStore() Store {
+	if p.cfg.SessionStore != StoreRedis {
+		ms := newMemoryStore()
+		_ = ms.Subscribe(context.Background(), p.handleEvent)
+		return ms
+	}
+	url := ""
+	prefix := cooler.KeyPrefix(nil)
+	if p.app.Config != nil {
+		url = strings.TrimSpace(p.app.Config.String(cooler.RedisURLKey))
+		prefix = cooler.KeyPrefix(p.app.Config)
+	}
+	return &lazyStore{open: func(ctx context.Context) (Store, error) {
+		if err := redisURLError(url); err != nil {
+			return nil, err
+		}
+		client, err := cooler.OpenRedis(ctx, url)
+		if err != nil {
+			return nil, err
+		}
+		rs := newRedisStore(client, prefix)
+		if err := rs.Subscribe(context.Background(), p.handleEvent); err != nil {
+			_ = client.Close()
+			return nil, err
+		}
+		return rs, nil
+	}}
+}
+
 // Handler returns the endpoint's handler (the Origin check, the body limit,
 // the bearer verification and the session router), for tests and hosts
 // that serve it outside the route table.
--- a/modules/pier/prm.go
+++ b/modules/pier/prm.go
@@ -0,0 +1,59 @@
+package pier
+
+import (
+	"net/http"
+	"net/url"
+
+	"git.golem15.com/golem15/summercms/modules/pact"
+	"github.com/modelcontextprotocol/go-sdk/auth"
+	"github.com/modelcontextprotocol/go-sdk/oauthex"
+)
+
+// PRMScopes is the scopes_supported list of the protected resource
+// metadata: the OAuth server's MCP scopes plus backend (D-21).
+var PRMScopes = []string{"read", "write", "ai", "offline_access", "backend"}
+
+// mountPRM serves the RFC 9728 protected resource metadata at
+// /.well-known/oauth-protected-resource and at
+// /.well-known/oauth-protected-resource<mcp.path>, through the host's
+// plain GET route kind. The authorization
+// server is the published verifier's resource origin when a verifier is
+// published, else app.url. A mcp.resource that differs from the verifier's
+// resource is logged as a warning at boot (Pitfall 8).
+func (p *Pier) mountPRM() error {
+	issuer := p.cfg.AppURL
+	if verifier, ok := p.app.Lookup[pact.MCPTokenVerifier](); ok && verifier != nil {
+		if res := verifier.MCPResource(); res != "" {
+			if u, err := url.Parse(res); err == nil && u.Scheme != "" && u.Host != "" {
+				issuer = u.Scheme + "://" + u.Host
+			}
+			if res != p.cfg.Resource {
+				logger().Warn("pier: mcp.resource differs from the verifier's resource; clients that send the verifier's resource will be refused",
+					"mcp_resource", p.cfg.Resource, "verifier_resource", res)
+			}
+		}
+	}
+	meta := &oauthex.ProtectedResourceMetadata{
+		Resource:               p.cfg.Resource,
+		ScopesSupported:        append([]string(nil), PRMScopes...),
+		BearerMethodsSupported: []string{"header"},
+	}
+	if issuer != "" {
+		meta.AuthorizationServers = []string{issuer}
+	}
+	h := nosniff(auth.ProtectedResourceMetadataHandler(meta))
+	for _, path := range []string{wellKnownPRM, wellKnownPRM + p.cfg.Path} {
+		if err := p.host.MountJSON(http.MethodGet, path, h); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// nosniff adds X-Content-Type-Options to a JSON document route.
+func nosniff(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-Content-Type-Options", "nosniff")
+		next.ServeHTTP(w, r)
+	})
+}
--- a/modules/pier/session.go
+++ b/modules/pier/session.go
@@ -10,6 +10,7 @@ import (
 	"io"
 	"net/http"
 	"slices"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -346,9 +347,43 @@ func (sr *sessionRouter) persist(ctx context.Context, ls *localSession) {
 	}
 }

-// serveStream serves the standalone GET stream.
+// serveStream serves the standalone GET stream and re-verifies its bearer
+// every recheck interval on a goroutine that holds the raw bearer only for
+// the stream's lifetime; a bearer that no longer resolves to the session's
+// binding ends the stream (Pitfall 10). The server's base context ends
+// every stream at shutdown.
 func (sr *sessionRouter) serveStream(w http.ResponseWriter, r *http.Request, ls *localSession) {
-	ls.transport.ServeHTTP(w, r)
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
+	raw := bearerOf(r)
+	go func() {
+		ticker := time.NewTicker(sr.p.recheck)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				id, err := sr.p.identities.resolve(ctx, raw)
+				if err != nil || id.Binding != ls.binding {
+					logSession(logger(), "stream_closed", ls.id, ls.binding)
+					cancel()
+					return
+				}
+				_ = sr.p.store.Touch(ctx, ls.id, sr.p.cfg.IdleTTL)
+			}
+		}
+	}()
+	ls.transport.ServeHTTP(w, r.WithContext(ctx))
+}
+
+// bearerOf reads the bearer of a request's Authorization header.
+func bearerOf(r *http.Request) string {
+	fields := strings.Fields(r.Header.Get("Authorization"))
+	if len(fields) != 2 || !strings.EqualFold(fields[0], "bearer") {
+		return ""
+	}
+	return fields[1]
 }

 // closeLocal closes and forgets this instance's copy of a session.
--- a/modules/pier/sessions_smoke_test.go
+++ b/modules/pier/sessions_smoke_test.go
@@ -0,0 +1,247 @@
+package pier_test
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/backpack"
+	"git.golem15.com/golem15/summercms/modules/pact"
+	"git.golem15.com/golem15/summercms/modules/party"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/modelcontextprotocol/go-sdk/mcp"
+)
+
+// Smoke tests of phase 19.7 plan 02 Task 3: Redis sessions shared by two
+// pier instances on one miniredis (rehydration, binding, DELETE, idle TTL,
+// the per-binding cap) and the pub/sub fan-out of subject events.
+
+// instance is one mounted pier with its app and test server.
+type instance struct {
+	app  *backpack.App
+	host *testHost
+	srv  *httptest.Server
+}
+
+// newInstance mounts pier on a fresh app sharing the miniredis and guard.
+func newInstance(t *testing.T, mr *miniredis.Miniredis, mcpYAML string, guard *acmeGuard) *instance {
+	t.Helper()
+	files := map[string]string{"mcp.yaml": mcpYAML}
+	if mr != nil {
+		files["redis.yaml"] = "url: redis://" + mr.Addr() + "/0\n"
+	}
+	app := pierAppWith(t, files, guard)
+	host, err := mountPier(t, app, []party.Plugin{acmePlugin{tools: []pact.MCPTool{echoTool()}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := httptest.NewServer(host)
+	t.Cleanup(srv.Close)
+	return &instance{app: app, host: host, srv: srv}
+}
+
+// rpcReply is one raw JSON-RPC exchange: the status, the session header
+// and the first message (an SSE data line or the JSON body).
+type rpcReply struct {
+	status  int
+	session string
+	body    string
+	header  http.Header
+}
+
+// rpc posts one JSON-RPC message with a bearer and an optional session id.
+func rpc(t *testing.T, ctx context.Context, url, token, session, body string) rpcReply {
+	t.Helper()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBufferString(body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("Authorization", "Bearer "+token)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json, text/event-stream")
+	req.Header.Set("Mcp-Protocol-Version", "2025-11-25")
+	if session != "" {
+		req.Header.Set("Mcp-Session-Id", session)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	reply := rpcReply{status: resp.StatusCode, session: resp.Header.Get("Mcp-Session-Id"), header: resp.Header}
+	if strings.HasPrefix(resp.Header.Get("Content-Type"), "text/event-stream") {
+		scanner := bufio.NewScanner(resp.Body)
+		scanner.Buffer(make([]byte, 1<<20), 1<<20)
+		for scanner.Scan() {
+			line := scanner.Text()
+			if strings.HasPrefix(line, "data: ") {
+				reply.body = strings.TrimPrefix(line, "data: ")
+				break
+			}
+		}
+		return reply
+	}
+	raw, _ := io.ReadAll(resp.Body)
+	reply.body = strings.TrimSpace(string(raw))
+	return reply
+}
+
+const initializeBody = `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"raw","version":"1"}}}`
+const initializedBody = `{"jsonrpc":"2.0","method":"notifications/initialized"}`
+const callBody = `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"acme_echo","arguments":{"id":9,"text":"hey"}}}`
+
+// openSession initialises a session on an instance and completes the
+// handshake, returning the session id.
+func openSession(t *testing.T, ctx context.Context, in *instance, token string) string {
+	t.Helper()
+	init := rpc(t, ctx, in.srv.URL+"/mcp", token, "", initializeBody)
+	if init.status != http.StatusOK || init.session == "" || !strings.Contains(init.body, `"protocolVersion":"2025-11-25"`) {
+		t.Fatalf("initialize = %d session %q body %s", init.status, init.session, init.body)
+	}
+	if done := rpc(t, ctx, in.srv.URL+"/mcp", token, init.session, initializedBody); done.status != http.StatusAccepted {
+		t.Fatalf("initialized = %d %s", done.status, done.body)
+	}
+	return init.session
+}
+
+func TestPierSessionsSmoke(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	mr := miniredis.RunT(t)
+	cfg := "enabled: false\nsessions:\n  store: redis\n  idle_ttl: 30m\n  max_per_binding: 2\n"
+	a := newInstance(t, mr, cfg, &acmeGuard{})
+	b := newInstance(t, mr, cfg, &acmeGuard{})
+
+	// Initialise on A, call on B: B rehydrates the session from Redis.
+	sid := openSession(t, ctx, a, "oas_at_test")
+	call := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_test", sid, callBody)
+	if call.status != http.StatusOK || !strings.Contains(call.body, `\"principal\":7`) || strings.Contains(call.body, `"isError":true`) {
+		t.Fatalf("call on B = %d %s", call.status, call.body)
+	}
+	keys := mr.Keys()
+	if !containsPrefix(keys, "acme:mcp:s:") || !containsPrefix(keys, "acme:mcp:b:oauth:c1:7") {
+		t.Fatalf("redis keys = %v, want acme:mcp:s:<id> and acme:mcp:b:oauth:c1:7", keys)
+	}
+
+	// Another binding presenting the same id is refused.
+	other := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_other", sid, callBody)
+	if other.status != http.StatusForbidden || !strings.Contains(other.body, "session_binding") {
+		t.Fatalf("other binding = %d %s", other.status, other.body)
+	}
+
+	// DELETE on B ends the session everywhere: A answers 404 next.
+	del, err := http.NewRequestWithContext(ctx, http.MethodDelete, b.srv.URL+"/mcp", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	del.Header.Set("Authorization", "Bearer oas_at_test")
+	del.Header.Set("Mcp-Session-Id", sid)
+	resp, err := http.DefaultClient.Do(del)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusNoContent {
+		t.Fatalf("DELETE on B = %d", resp.StatusCode)
+	}
+	if gone := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", sid, callBody); gone.status != http.StatusNotFound {
+		t.Fatalf("call on A after DELETE = %d %s", gone.status, gone.body)
+	}
+
+	// An idle session expires: after the TTL the id is unknown.
+	expiring := openSession(t, ctx, a, "oas_at_test")
+	mr.FastForward(31 * time.Minute)
+	if expired := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", expiring, callBody); expired.status != http.StatusNotFound {
+		t.Fatalf("call after the idle TTL = %d %s", expired.status, expired.body)
+	}
+
+	// max_per_binding + 1 initialisations answer 429.
+	openSession(t, ctx, a, "oas_at_test")
+	openSession(t, ctx, b, "oas_at_test")
+	third := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", "", initializeBody)
+	if third.status != http.StatusTooManyRequests || !strings.Contains(third.body, "too_many_sessions") {
+		t.Fatalf("third session = %d %s", third.status, third.body)
+	}
+}
+
+func containsPrefix(keys []string, prefix string) bool {
+	for _, k := range keys {
+		if strings.HasPrefix(k, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+func TestPierFanOutSmoke(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	mr := miniredis.RunT(t)
+	guard := &acmeGuard{}
+	cfg := "enabled: false\nsessions:\n  store: redis\n"
+	a := newInstance(t, mr, cfg, guard)
+	b := newInstance(t, mr, cfg, guard)
+
+	changed := make(chan struct{}, 4)
+	client := mcp.NewClient(&mcp.Implementation{Name: "acme-test", Version: "1"}, &mcp.ClientOptions{
+		ToolListChangedHandler: func(context.Context, *mcp.ToolListChangedRequest) {
+			select {
+			case changed <- struct{}{}:
+			default:
+			}
+		},
+	})
+	transport := &mcp.StreamableClientTransport{
+		Endpoint:   b.srv.URL + "/mcp",
+		HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}},
+	}
+	cs, err := client.Connect(ctx, transport, &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"})
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer cs.Close()
+	list, err := cs.ListTools(ctx, nil)
+	if err != nil || len(list.Tools) != 1 {
+		t.Fatalf("tools/list = %+v, %v", list, err)
+	}
+
+	// The guard now denies the tool; a permissions event fired on A reaches
+	// the session on B as notifications/tools/list_changed.
+	guard.deny.Store(true)
+	if err := a.app.Events.Fire(ctx, &pact.MCPSubjectChanged{Guard: "acme", UserID: 7, Kind: pact.MCPSubjectPermissions}); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-changed:
+	case <-time.After(10 * time.Second):
+		t.Fatal("no tools/list_changed within 10 s")
+	}
+	list, err = cs.ListTools(ctx, nil)
+	if err != nil || len(list.Tools) != 0 {
+		t.Fatalf("tools/list after the permission change = %+v, %v", list, err)
+	}
+
+	// A revocation event ends the session: the next call is refused.
+	if err := a.app.Events.Fire(ctx, &pact.MCPSubjectChanged{Guard: "acme", UserID: 7, Kind: pact.MCPSubjectRevoked}); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		if _, err := cs.ListTools(ctx, nil); err != nil {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("the session survived the revocation event")
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	if raw := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_test", cs.ID(), callBody); raw.status != http.StatusNotFound {
+		t.Fatalf("call after revocation = %d %s", raw.status, raw.body)
+	}
+}
--- a/modules/pier/stateless.go
+++ b/modules/pier/stateless.go
@@ -0,0 +1,174 @@
+package pier
+
+import (
+	"bytes"
+	"container/list"
+	"errors"
+	"io"
+	"net/http"
+	"slices"
+	"sync"
+
+	"github.com/modelcontextprotocol/go-sdk/mcp"
+)
+
+// serverFeatures is what applyDiff needs from an SDK server.
+type serverFeatures interface {
+	AddTool(t *mcp.Tool, h mcp.ToolHandler)
+	RemoveTools(names ...string)
+	AddPrompt(p *mcp.Prompt, h mcp.PromptHandler)
+	RemovePrompts(names ...string)
+}
+
+// statelessEntry is one per-identity server of the stateless path, keyed
+// by binding, with the names it holds for fan-out diffs.
+type statelessEntry struct {
+	binding string
+	server  *mcp.Server
+	elem    *list.Element
+
+	mu       sync.Mutex
+	identity Identity
+	tools    map[string]bool
+	prompts  map[string]bool
+}
+
+// statelessPath serves clients of protocol 2026-07-28 and later (D-32,
+// P-08): the SDK's stateless handler over an in-process LRU of
+// per-identity servers (capacity mcp.sessions.max_per_binding * 16),
+// evicted on revocation, mutated on fan-out events so subscriptions/listen
+// clients receive the list changes. Every request is authorized afresh by
+// the bearer middleware in front of it.
+type statelessPath struct {
+	p        *Pier
+	handler  http.Handler
+	capacity int
+
+	mu      sync.Mutex
+	entries map[string]*statelessEntry
+	order   *list.List // front = most recently used
+}
+
+// statelessCapacityFactor multiplies mcp.sessions.max_per_binding into the
+// LRU capacity.
+const statelessCapacityFactor = 16
+
+func newStatelessPath(p *Pier) *statelessPath {
+	sp := &statelessPath{
+		p:        p,
+		capacity: p.cfg.MaxPerBinding * statelessCapacityFactor,
+		entries:  map[string]*statelessEntry{},
+		order:    list.New(),
+	}
+	sp.handler = mcp.NewStreamableHTTPHandler(sp.getServer, &mcp.StreamableHTTPOptions{
+		Stateless: true,
+		// pier runs behind a reverse proxy and checks Origin itself; the
+		// SDK's loopback rule would refuse every proxied request of a
+		// binary listening on 127.0.0.1.
+		DisableLocalhostProtection: true,
+		MaxRequestBodyBytes:        maxRequestBody,
+	})
+	return sp
+}
+
+func (sp *statelessPath) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	sp.handler.ServeHTTP(w, r)
+}
+
+// getServer returns the per-identity server of the request's binding,
+// building it on a miss and refreshing the identity on a hit.
+func (sp *statelessPath) getServer(r *http.Request) *mcp.Server {
+	id, ok := identityFromRequest(r)
+	if !ok {
+		return nil
+	}
+	sp.mu.Lock()
+	defer sp.mu.Unlock()
+	if e, ok := sp.entries[id.Binding]; ok {
+		sp.order.MoveToFront(e.elem)
+		e.mu.Lock()
+		e.identity = id
+		e.mu.Unlock()
+		return e.server
+	}
+	srv, tools, prompts := sp.p.newServer(r.Context(), id, nil)
+	e := &statelessEntry{binding: id.Binding, server: srv, identity: id, tools: tools, prompts: prompts}
+	e.elem = sp.order.PushFront(e)
+	sp.entries[id.Binding] = e
+	for sp.order.Len() > sp.capacity {
+		last := sp.order.Back()
+		if last == nil {
+			break
+		}
+		old := last.Value.(*statelessEntry)
+		sp.order.Remove(last)
+		delete(sp.entries, old.binding)
+	}
+	return srv
+}
+
+// matching lists the entries whose identity includes the subject.
+func (sp *statelessPath) matching(guard string, userID uint) []*statelessEntry {
+	sp.mu.Lock()
+	defer sp.mu.Unlock()
+	var out []*statelessEntry
+	for _, e := range sp.entries {
+		e.mu.Lock()
+		hit := e.identity.has(guard, userID)
+		e.mu.Unlock()
+		if hit {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+// evict drops the entries of a subject (a revocation or deletion) and
+// closes their sessions, so a subscriptions/listen stream ends.
+func (sp *statelessPath) evict(guard string, userID uint) {
+	for _, e := range sp.matching(guard, userID) {
+		sp.mu.Lock()
+		if cur, ok := sp.entries[e.binding]; ok && cur == e {
+			sp.order.Remove(e.elem)
+			delete(sp.entries, e.binding)
+		}
+		sp.mu.Unlock()
+		for ss := range e.server.Sessions() {
+			_ = ss.Close()
+		}
+	}
+}
+
+// route sends a request to the stateless path when its protocol version
+// header is 2026-07-28 or later, or when the first JSON-RPC method of a
+// POST is server/discover; everything else goes to the stateful router
+// (P-08).
+func (p *Pier) route(w http.ResponseWriter, r *http.Request) {
+	if p.stateless == nil {
+		p.sessions.ServeHTTP(w, r)
+		return
+	}
+	if pv := r.Header.Get(protocolVersionHeader); pv >= protocolVersion202607 {
+		p.stateless.ServeHTTP(w, r)
+		return
+	}
+	if r.Method == http.MethodPost && r.Body != nil {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			var mbe *http.MaxBytesError
+			if errors.As(err, &mbe) {
+				http.Error(w, "request body too large", http.StatusRequestEntityTooLarge)
+				return
+			}
+			http.Error(w, "failed to read body", http.StatusBadRequest)
+			return
+		}
+		_ = r.Body.Close()
+		r.Body = io.NopCloser(bytes.NewReader(body))
+		if slices.Contains(peekMethods(body), methodDiscover) {
+			p.stateless.ServeHTTP(w, r)
+			return
+		}
+	}
+	p.sessions.ServeHTTP(w, r)
+}
--- a/modules/pier/stateless_smoke_test.go
+++ b/modules/pier/stateless_smoke_test.go
@@ -0,0 +1,80 @@
+package pier_test
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/pact"
+	"git.golem15.com/golem15/summercms/modules/party"
+	"github.com/modelcontextprotocol/go-sdk/mcp"
+)
+
+// Smoke test of phase 19.7 plan 02 Task 3: the hybrid protocol (D-32). A
+// client at the SDK's newest protocol negotiates 2026-07-28 statelessly
+// (no Mcp-Session-Id) and calls the tool; a client pinned to 2025-11-25
+// gets a session on the stateful path; both see the same per-identity list.
+func TestPierStatelessSmoke(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	app := pierApp(t, "enabled: false\n")
+	host, err := mountPier(t, app, []party.Plugin{acmePlugin{tools: []pact.MCPTool{echoTool()}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := httptest.NewServer(host)
+	defer srv.Close()
+	transport := func() *mcp.StreamableClientTransport {
+		return &mcp.StreamableClientTransport{
+			Endpoint:   srv.URL + "/mcp",
+			HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}},
+		}
+	}
+
+	newest := mcp.NewClient(&mcp.Implementation{Name: "newest", Version: "1"}, nil)
+	stateless, err := newest.Connect(ctx, transport(), nil)
+	if err != nil {
+		t.Fatalf("stateless connect: %v", err)
+	}
+	defer stateless.Close()
+	if stateless.ID() != "" {
+		t.Fatalf("the newest client got a session id %q, want none", stateless.ID())
+	}
+	if v := stateless.InitializeResult().ProtocolVersion; v != "2026-07-28" {
+		t.Fatalf("negotiated %q, want 2026-07-28", v)
+	}
+	res, err := stateless.CallTool(ctx, &mcp.CallToolParams{Name: "acme_echo", Arguments: map[string]any{"id": 3, "text": "yo"}})
+	if err != nil || res.IsError {
+		t.Fatalf("stateless tools/call = %+v, %v", res, err)
+	}
+	if structured, _ := res.StructuredContent.(map[string]any); structured["principal"] != 7.0 || structured["id"] != 3.0 {
+		t.Fatalf("stateless structuredContent = %+v", res.StructuredContent)
+	}
+
+	legacy := mcp.NewClient(&mcp.Implementation{Name: "legacy", Version: "1"}, nil)
+	stateful, err := legacy.Connect(ctx, transport(), &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"})
+	if err != nil {
+		t.Fatalf("stateful connect: %v", err)
+	}
+	defer stateful.Close()
+	if stateful.ID() == "" {
+		t.Fatal("the 2025-11-25 client got no session id")
+	}
+	if v := stateful.InitializeResult().ProtocolVersion; v != "2025-11-25" {
+		t.Fatalf("negotiated %q, want 2025-11-25", v)
+	}
+
+	a, err := stateless.ListTools(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := stateful.ListTools(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(a.Tools) != 1 || len(b.Tools) != 1 || a.Tools[0].Name != b.Tools[0].Name || a.Tools[0].Name != "acme_echo" {
+		t.Fatalf("lists differ: stateless %+v, stateful %+v", a.Tools, b.Tools)
+	}
+}
--- a/modules/pier/store_redis.go
+++ b/modules/pier/store_redis.go
@@ -0,0 +1,247 @@
+package pier
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// Redis key and channel layout under cooler's application prefix.
+const (
+	redisSessionKey = "mcp:s:"     // <prefix>mcp:s:<id> holds the JSON record
+	redisBindingKey = "mcp:b:"     // <prefix>mcp:b:<binding> is the set of its session ids
+	redisEventsKey  = "mcp:events" // <prefix>mcp:events carries JSON events
+)
+
+// redisStore keeps session records in Redis with PX TTLs and fans events
+// out through pub/sub, so several instances share sessions and react to
+// each other's events.
+type redisStore struct {
+	client *redis.Client
+	prefix string
+}
+
+func newRedisStore(client *redis.Client, prefix string) *redisStore {
+	return &redisStore{client: client, prefix: prefix}
+}
+
+func (s *redisStore) sessionKey(id string) string { return s.prefix + redisSessionKey + id }
+
+func (s *redisStore) bindingKey(binding string) string { return s.prefix + redisBindingKey + binding }
+
+func (s *redisStore) channel() string { return s.prefix + redisEventsKey }
+
+func (s *redisStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error {
+	raw, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	pipe := s.client.TxPipeline()
+	pipe.Set(ctx, s.sessionKey(id), raw, ttl)
+	pipe.SAdd(ctx, s.bindingKey(rec.Binding), id)
+	pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl)
+	_, err = pipe.Exec(ctx)
+	return err
+}
+
+func (s *redisStore) Get(ctx context.Context, id string) (Record, bool, error) {
+	raw, err := s.client.Get(ctx, s.sessionKey(id)).Bytes()
+	if errors.Is(err, redis.Nil) {
+		return Record{}, false, nil
+	}
+	if err != nil {
+		return Record{}, false, err
+	}
+	var rec Record
+	if err := json.Unmarshal(raw, &rec); err != nil {
+		return Record{}, false, err
+	}
+	return rec, true, nil
+}
+
+// Touch slides the record's TTL and its binding set's with it.
+func (s *redisStore) Touch(ctx context.Context, id string, ttl time.Duration) error {
+	rec, ok, err := s.Get(ctx, id)
+	if err != nil || !ok {
+		return err
+	}
+	pipe := s.client.TxPipeline()
+	pipe.PExpire(ctx, s.sessionKey(id), ttl)
+	pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl)
+	_, err = pipe.Exec(ctx)
+	return err
+}
+
+func (s *redisStore) Delete(ctx context.Context, id string) error {
+	rec, ok, err := s.Get(ctx, id)
+	if err != nil {
+		return err
+	}
+	pipe := s.client.TxPipeline()
+	pipe.Del(ctx, s.sessionKey(id))
+	if ok {
+		pipe.SRem(ctx, s.bindingKey(rec.Binding), id)
+	}
+	_, err = pipe.Exec(ctx)
+	return err
+}
+
+// Count answers the live sessions of a binding, pruning members whose
+// record has expired from the set.
+func (s *redisStore) Count(ctx context.Context, binding string) (int, error) {
+	members, err := s.client.SMembers(ctx, s.bindingKey(binding)).Result()
+	if err != nil {
+		return 0, err
+	}
+	n := 0
+	for _, id := range members {
+		exists, err := s.client.Exists(ctx, s.sessionKey(id)).Result()
+		if err != nil {
+			return 0, err
+		}
+		if exists > 0 {
+			n++
+			continue
+		}
+		_ = s.client.SRem(ctx, s.bindingKey(binding), id).Err()
+	}
+	return n, nil
+}
+
+func (s *redisStore) Publish(ctx context.Context, ev Event) error {
+	raw, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	return s.client.Publish(ctx, s.channel(), raw).Err()
+}
+
+// Subscribe delivers every event published on the channel to fn, on its
+// own goroutine, until ctx ends.
+func (s *redisStore) Subscribe(ctx context.Context, fn func(Event)) error {
+	sub := s.client.Subscribe(ctx, s.channel())
+	if _, err := sub.Receive(ctx); err != nil {
+		_ = sub.Close()
+		return err
+	}
+	go func() {
+		defer sub.Close()
+		ch := sub.Channel()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case msg, ok := <-ch:
+				if !ok {
+					return
+				}
+				var ev Event
+				if err := json.Unmarshal([]byte(msg.Payload), &ev); err != nil {
+					continue
+				}
+				fn(ev)
+			}
+		}
+	}()
+	return nil
+}
+
+// lazyStore opens the Redis store on first use, so nothing dials at mount
+// (route:list and tests never dial) and a missing redis.url is a 503 on the
+// first request, logged once, not a boot failure (cooler's rule).
+type lazyStore struct {
+	open   func(ctx context.Context) (Store, error)
+	mu     sync.Mutex
+	store  Store
+	warned atomic.Bool
+}
+
+// get returns the store, opening it on the first call and retrying on a
+// later call after a failed open.
+func (l *lazyStore) get(ctx context.Context) (Store, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.store != nil {
+		return l.store, nil
+	}
+	s, err := l.open(ctx)
+	if err != nil {
+		if !l.warned.Swap(true) {
+			logger().Error("pier: session store could not be opened", "error", err)
+		}
+		return nil, err
+	}
+	l.store = s
+	return s, nil
+}
+
+func (l *lazyStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error {
+	s, err := l.get(ctx)
+	if err != nil {
+		return err
+	}
+	return s.Put(ctx, id, rec, ttl)
+}
+
+func (l *lazyStore) Get(ctx context.Context, id string) (Record, bool, error) {
+	s, err := l.get(ctx)
+	if err != nil {
+		return Record{}, false, err
+	}
+	return s.Get(ctx, id)
+}
+
+func (l *lazyStore) Touch(ctx context.Context, id string, ttl time.Duration) error {
+	s, err := l.get(ctx)
+	if err != nil {
+		return err
+	}
+	return s.Touch(ctx, id, ttl)
+}
+
+func (l *lazyStore) Delete(ctx context.Context, id string) error {
+	s, err := l.get(ctx)
+	if err != nil {
+		return err
+	}
+	return s.Delete(ctx, id)
+}
+
+func (l *lazyStore) Count(ctx context.Context, binding string) (int, error) {
+	s, err := l.get(ctx)
+	if err != nil {
+		return 0, err
+	}
+	return s.Count(ctx, binding)
+}
+
+func (l *lazyStore) Publish(ctx context.Context, ev Event) error {
+	s, err := l.get(ctx)
+	if err != nil {
+		return err
+	}
+	return s.Publish(ctx, ev)
+}
+
+func (l *lazyStore) Subscribe(ctx context.Context, fn func(Event)) error {
+	s, err := l.get(ctx)
+	if err != nil {
+		return err
+	}
+	return s.Subscribe(ctx, fn)
+}
+
+// redisURLError names the key a redis store needs.
+func redisURLError(url string) error {
+	if strings.TrimSpace(url) == "" {
+		return fmt.Errorf("pier: redis.url is required when %s is %s (set it in config/redis.yaml or SUMMER_REDIS__URL)", SessionStoreKey, StoreRedis)
+	}
+	return nil
+}
--- a/modules/pier/tracer_smoke_test.go
+++ b/modules/pier/tracer_smoke_test.go
@@ -12,6 +12,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -106,11 +107,14 @@ func (p acmePlugin) MCP() pact.MCP {
 }

 // acmeGuard is the fake guard provider for "acme": token-guarded by
-// acme.auth, allows everything, dispatches as the subject's user id.
-type acmeGuard struct{}
+// acme.auth, allows everything unless deny is set, dispatches as the
+// subject's user id.
+type acmeGuard struct {
+	deny atomic.Bool
+}

-func (acmeGuard) MCPGuardName() string { return "acme" }
-func (acmeGuard) MCPTokenGuarded(mw []string) bool {
+func (*acmeGuard) MCPGuardName() string { return "acme" }
+func (*acmeGuard) MCPTokenGuarded(mw []string) bool {
 	for _, m := range mw {
 		if m == "acme.auth" {
 			return true
@@ -118,24 +122,28 @@ func (acmeGuard) MCPTokenGuarded(mw []string) bool {
 	}
 	return false
 }
-func (acmeGuard) MCPRequirements([]string) (pact.MCPRequirements, error) {
+func (*acmeGuard) MCPRequirements([]string) (pact.MCPRequirements, error) {
 	return pact.MCPRequirements{}, nil
 }
-func (acmeGuard) MCPAllows(context.Context, pact.MCPSubject, pact.MCPRequirements, []string) (bool, error) {
-	return true, nil
+func (g *acmeGuard) MCPAllows(context.Context, pact.MCPSubject, pact.MCPRequirements, []string) (bool, error) {
+	return !g.deny.Load(), nil
 }
-func (acmeGuard) MCPDispatch(_ context.Context, subject pact.MCPSubject, _ string) (pact.MCPDispatch, error) {
+func (*acmeGuard) MCPDispatch(_ context.Context, subject pact.MCPSubject, _ string) (pact.MCPDispatch, error) {
 	return pact.MCPDispatch{Principal: &bouncer.Principal{ID: subject.UserID}, Guards: []string{"acme"}}, nil
 }

-// acmeVerifier accepts oas_at_test for user 7, client c1, read and write.
+// acmeVerifier accepts oas_at_test for user 7, client c1, read and write,
+// and oas_at_other for user 8, client c2 (another binding).
 type acmeVerifier struct{}

 func (acmeVerifier) MCPVerifyToken(_ context.Context, raw string) (pact.MCPGrant, error) {
-	if raw != "oas_at_test" {
-		return pact.MCPGrant{}, pact.ErrMCPTokenInvalid
+	switch raw {
+	case "oas_at_test":
+		return pact.MCPGrant{UserID: 7, ClientID: "c1", TokenID: 1, Scopes: []string{"read", "write"}, ExpiresAt: time.Now().Add(time.Hour)}, nil
+	case "oas_at_other":
+		return pact.MCPGrant{UserID: 8, ClientID: "c2", TokenID: 2, Scopes: []string{"read", "write"}, ExpiresAt: time.Now().Add(time.Hour)}, nil
 	}
-	return pact.MCPGrant{UserID: 7, ClientID: "c1", TokenID: 1, Scopes: []string{"read", "write"}, ExpiresAt: time.Now().Add(time.Hour)}, nil
+	return pact.MCPGrant{}, pact.ErrMCPTokenInvalid
 }

 func (acmeVerifier) MCPResource() string { return "http://acme.test/mcp" }
@@ -196,16 +204,26 @@ func (h *testHost) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	http.NotFound(w, r)
 }

-// pierApp builds an app with mcp.yaml from the given body, publishes the
-// fake verifier and guard registry, builds the router with the MCP flag
-// off (so no tag is needed) and then enables it for pier.Mount.
+// pierApp builds an app with mcp.yaml from the given body and publishes
+// the fake verifier and a guard registry with a fresh acme guard.
 func pierApp(t *testing.T, mcpYAML string) *backpack.App {
+	t.Helper()
+	return pierAppWith(t, map[string]string{"mcp.yaml": mcpYAML}, &acmeGuard{})
+}
+
+// pierAppWith builds an app from extra config files (app.yaml and
+// http.yaml have defaults) and publishes the fake verifier and the given
+// guard provider; the router is built with the MCP flag off (so no tag is
+// needed) and the flag is enabled for pier.Mount afterwards.
+func pierAppWith(t *testing.T, extra map[string]string, guard *acmeGuard) *backpack.App {
 	t.Helper()
 	dir := t.TempDir()
 	files := map[string]string{
 		"app.yaml":  "name: acme\nurl: http://acme.test\n",
 		"http.yaml": "body_limits:\n  default_bytes: 1048576\n  upload_bytes: 1048576\n",
-		"mcp.yaml":  mcpYAML,
+	}
+	for name, body := range extra {
+		files[name] = body
 	}
 	for name, body := range files {
 		if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o644); err != nil {
@@ -221,7 +239,7 @@ func pierApp(t *testing.T, mcpYAML string) *backpack.App {
 		t.Fatal(err)
 	}
 	reg := pact.NewMCPGuards()
-	if err := reg.Register(acmeGuard{}); err != nil {
+	if err := reg.Register(guard); err != nil {
 		t.Fatal(err)
 	}
 	if err := app.Publish(reg); err != nil {
@@ -290,7 +308,9 @@ func TestPierTracerSmoke(t *testing.T) {
 		Endpoint:   srv.URL + "/mcp",
 		HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}},
 	}
-	cs, err := client.Connect(ctx, transport, nil)
+	// Pinned to 2025-11-25: the stateful path with sessions (the newest
+	// protocol is served statelessly, see TestPierStatelessSmoke).
+	cs, err := client.Connect(ctx, transport, &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"})
 	if err != nil {
 		t.Fatalf("connect: %v", err)
 	}