fix(19.7): RA-WR-04 LRU eviction spares stateless servers that are serving

Each stateless request now carries an entry-use record on its context;
getServer counts the entry it hands out as active (under the path lock)
and ServeHTTP releases it when the request ends, so a subscriptions/listen
stream keeps its entry active for its whole life. Capacity eviction walks
the LRU from the back and drops only idle entries, so it no longer ends a
quiet listener or cancels a tool call in flight, and a revocation still
finds every open stream. While every entry is busy the LRU stays above
capacity until requests end.

TestStatelessFanOut's LRU subtest now asserts that the listener's server is
spared when the binding past capacity arrives (the oldest idle server goes
instead) and that a revocation then still ends the stream.

modules/pier/README.md

@@ -40,7 +40,7 @@ Use cases:
 - 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`); Mount never waits for Redis: a background warm-up opens the client from `redis.url` with a backoff (one second doubling to thirty, each failure logged) and subscribes to the event channel, so an instance that serves only stateless clients still receives fan-out, and the first request on any path also starts the open without waiting for it; while the store is down stateful requests answer 503 and stateless requests are served.
 - 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. A revocation naming `TokenID` ends only the sessions whose subject under the event's guard is that user with that token id and no client id (one personal token); one naming `ClientID` ends only that OAuth client's sessions of the user; one naming neither, and every deletion, ends all sessions of the subject.
 - Client address: before resolving a bearer, pier puts the client address on the resolvers' context with `towel.WithClientIP`, taken from the host when it offers one (surf applies `http.trusted_proxies`) or from the request's `RemoteAddr`, so a resolver can log a failed attempt with it.
-- 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. A server evicted from the LRU closes its sessions, and a stateless `subscriptions/listen` stream re-verifies its bearer every minute and ends when it no longer resolves to its binding, as the stateful GET stream does.
+- 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. Past capacity the least recently used idle server is evicted (and its sessions closed); a server that is serving a request, a `subscriptions/listen` stream included, is never evicted, so capacity never ends a quiet listener or a tool call in flight, and the LRU stays above capacity while every server is busy. A stateless `subscriptions/listen` stream re-verifies its bearer every minute and ends when it no longer resolves to its binding, as the stateful GET stream does.
 - Widening on return (A-WR-04): when a binding comes back with a different identity (more or fewer scopes, a backend subject added or dropped), its stateful session or stateless server is re-filtered on that request, so newly allowed tools and prompts appear at once (with `list_changed`) instead of waiting for an event. Token ids play no part in this, so an access-token rotation alone re-filters nothing. A stateless server is shared by every live token of its binding, so a return with less reach only adds there and never removes: the receiving middleware hides, per request, what that request's identity may not use, and a call of a wider token is never cut off by a narrower one. A guard provider error during this re-filtering is not a refusal: the cached server stays as it was and the next request retries (logged as `pier: widening deferred`); a fan-out recompute that meets one hides the failing items and leaves the server to be re-filtered by the binding's next request.
 - 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 prompt declared with `Writes` (it directs tools that are not read-only) sits under the read/write ceiling and is hidden from an identity without `write`.

modules/pier/stateless.go

@@ -28,6 +28,10 @@ type statelessEntry struct {
 	binding string
 	server  *mcp.Server
 	elem    *list.Element
+	// active counts the requests being served on server (a
+	// subscriptions/listen stream for its whole life); guarded by
+	// statelessPath.mu. The LRU never evicts an active entry (RA-WR-04).
+	active int
 
 	// mu serializes re-filtering and guards fingerprint, widened, tools and
 	// prompts (see localSession); identity has its own box.
@@ -61,6 +65,16 @@ type statelessPath struct {
 // LRU capacity.
 const statelessCapacityFactor = 16
 
+// entryUseKey keys the entryUse of one stateless request on its context.
+type entryUseKey struct{}
+
+// entryUse lists the entries one stateless request is served by, so
+// ServeHTTP releases them when the request ends; guarded by
+// statelessPath.mu.
+type entryUse struct {
+	entries []*statelessEntry
+}
+
 func newStatelessPath(p *Pier) *statelessPath {
 	sp := &statelessPath{
 		p:        p,
@@ -84,6 +98,9 @@ func newStatelessPath(p *Pier) *statelessPath {
 // every recheck interval and ends when the bearer no longer resolves to
 // the binding that opened it (A-WR-06).
 func (sp *statelessPath) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	use := &entryUse{}
+	r = r.WithContext(context.WithValue(r.Context(), entryUseKey{}, use))
+	defer sp.release(use)
 	if r.Method != http.MethodPost || r.Body == nil {
 		sp.handler.ServeHTTP(w, r)
 		return
@@ -135,7 +152,7 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server {
 	if !ok {
 		return nil
 	}
-	if e := sp.hit(id.Binding); e != nil {
+	if e := sp.hit(r, id.Binding); e != nil {
 		sp.refresh(r.Context(), e, id)
 		return e.server
 	}
@@ -147,34 +164,39 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server {
 	sp.mu.Lock()
 	if cur, ok := sp.entries[id.Binding]; ok {
 		sp.order.MoveToFront(cur.elem)
+		sp.track(r, cur)
 		sp.mu.Unlock()
 		sp.refresh(r.Context(), cur, id)
 		return cur.server
 	}
 	e.elem = sp.order.PushFront(e)
 	sp.entries[id.Binding] = e
+	sp.track(r, e)
+	// Past capacity the least recently used idle entries go. An entry that
+	// is serving a request (a quiet subscriptions/listen stream, a tool call
+	// in flight) is never evicted, so capacity never cuts a stream or a call
+	// off and a revocation still finds every open stream (A-WR-06, RA-WR-04);
+	// while every entry is busy the LRU stays above capacity.
 	var evicted []*statelessEntry
-	for sp.order.Len() > sp.capacity {
-		last := sp.order.Back()
-		if last == nil {
-			break
+	for el := sp.order.Back(); el != nil && sp.order.Len() > sp.capacity; {
+		prev := el.Prev()
+		if old := el.Value.(*statelessEntry); old.active == 0 {
+			sp.order.Remove(el)
+			delete(sp.entries, old.binding)
+			evicted = append(evicted, old)
 		}
-		old := last.Value.(*statelessEntry)
-		sp.order.Remove(last)
-		delete(sp.entries, old.binding)
-		evicted = append(evicted, old)
+		el = prev
 	}
 	sp.mu.Unlock()
-	// An evicted server's listen streams end with it, so none outlives the
-	// entry a later revocation would look for (A-WR-06).
 	for _, old := range evicted {
 		closeSessions(old.server)
 	}
 	return srv
 }
 
-// hit answers the entry of a binding, most recently used now, or nil.
-func (sp *statelessPath) hit(binding string) *statelessEntry {
+// hit answers the entry of a binding, most recently used now and counted as
+// serving r, or nil.
+func (sp *statelessPath) hit(r *http.Request, binding string) *statelessEntry {
 	sp.mu.Lock()
 	defer sp.mu.Unlock()
 	e, ok := sp.entries[binding]
@@ -182,9 +204,32 @@ func (sp *statelessPath) hit(binding string) *statelessEntry {
 		return nil
 	}
 	sp.order.MoveToFront(e.elem)
+	sp.track(r, e)
 	return e
 }
 
+// track counts e as serving r until ServeHTTP releases r's entries; the
+// caller holds sp.mu. A request reaching getServer without ServeHTTP's
+// entryUse is not counted.
+func (sp *statelessPath) track(r *http.Request, e *statelessEntry) {
+	use, _ := r.Context().Value(entryUseKey{}).(*entryUse)
+	if use == nil || slices.Contains(use.entries, e) {
+		return
+	}
+	e.active++
+	use.entries = append(use.entries, e)
+}
+
+// release ends the request's use of its entries.
+func (sp *statelessPath) release(use *entryUse) {
+	sp.mu.Lock()
+	for _, e := range use.entries {
+		e.active--
+	}
+	use.entries = nil
+	sp.mu.Unlock()
+}
+
 // refresh stores the request's identity on an entry and, when the
 // identity's fingerprint is new to the entry, adds the tools and prompts
 // it allows (A-WR-04). It never removes any (RA-WR-02): one server serves

modules/pier/stateless_fanout_test.go

@@ -91,8 +91,9 @@ func waitEnded(t *testing.T, ended <-chan struct{}, within time.Duration, what s
 // A-WR-06): an instance that served only stateless requests receives a
 // revocation published by the other instance through Redis and evicts its
 // entry; a subscriptions/listen stream ends within the re-check interval
-// once its bearer stops resolving; and an LRU eviction at capacity closes
-// the evicted server's listen stream.
+// once its bearer stops resolving; and an LRU eviction at capacity spares a
+// server with an open listen stream (RA-WR-04), evicting the least recently
+// used idle one, while a revocation still ends that stream.
 func TestStatelessFanOut(t *testing.T) {
 	plugin := acmePlugin{tools: []pact.MCPTool{echoTool(), showTool()}}
 
@@ -138,7 +139,7 @@ func TestStatelessFanOut(t *testing.T) {
 		}
 	})
 
-	t.Run("an LRU eviction closes the evicted server's listen stream", func(t *testing.T) {
+	t.Run("an LRU eviction spares a server with a listen stream", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 		defer cancel()
 		const capacity = 16 // max_per_binding 1 times the LRU factor
@@ -164,10 +165,20 @@ func TestStatelessFanOut(t *testing.T) {
 		if n := in.pier().StatelessServers(); n != capacity {
 			t.Fatalf("stateless servers at capacity = %d, want %d", n, capacity)
 		}
+		// One binding past capacity: the listener's server is the least
+		// recently used, but it is serving the stream, so the oldest idle
+		// server goes instead.
 		listNames(t, ctx, sdkClient(t, ctx, in.url(), fmt.Sprintf("oas_at_lru_%d", capacity-1), clientOpts{version: "newest"}))
-		waitEnded(t, ended, 5*time.Second, "LRU eviction")
+		select {
+		case <-ended:
+			t.Fatal("the LRU eviction ended a live listen stream")
+		case <-time.After(300 * time.Millisecond):
+		}
 		if n := in.pier().StatelessServers(); n != capacity {
 			t.Fatalf("stateless servers after the eviction = %d, want %d", n, capacity)
 		}
+		// The spared entry is still where a revocation looks for it.
+		fireSubjectChanged(t, in.app, "acme", 20, pact.MCPSubjectRevoked)
+		waitEnded(t, ended, 5*time.Second, "revocation after the eviction")
 	})
 }