[mcollina]

Problem

On the legacy Streamable HTTP + SSE transport, the per-session topic mcp/session/<id>/message had an inconsistent lifecycle:

• Legacy SSE sessions stop receiving messages after their first GET stream closes, even with Last-Event-ID #327: only createSession subscribed, and the last stream closing unsubscribed. A later GET never subscribed again, so after the first stream closed, messages were neither delivered nor stored in history, while mcpSendToSession still returned true.
• Redis: a legacy SSE stream on another instance never receives per-session messages #328: with Redis, a GET on an instance other than the one that ran initialize never subscribed, so it never received per-session messages. A DELETE on A did not close B's stream.
• Redis backend leaks a broker subscription for every expired legacy session #329: with Redis, every session subscribed at initialize, and the stale-session sweep ran only for MemorySessionStore, so expired sessions leaked a broker subscription and a localStreams entry. On top of that, RedisMessageBroker.unsubscribe used mqemitter's base removeAllListeners, which never sends UNSUBSCRIBE. Even explicit unsubscribes leaked the channel in Redis.
• Resource unsubscribe handlers are not reference-counted across eras and never run on legacy session expiry #337 (main-branch part only): when a legacy session was DELETEd or expired, the resource unsubscribe handler never ran, so watchers started by the subscribe handler leaked.

Design

The obvious fix is "subscribe wherever a stream attaches and keep it for the session's lifetime". That doesn't work with several instances on its own. History was written by the subscriber when it had no local stream, so with A (creator) and B (stream holder) both subscribed, every message would be stored twice. So this PR moves history writes to the sender.

1. Sender stores, then publishes (src/session-topic.ts). mcpSendToSession (and so elicitation and task status updates) writes the message to history, then publishes { eventId, message } on the session topic. History is exactly-once and no longer depends on any subscription. If the store fails, mcpSendToSession now returns false instead of true.
2. Subscribe exactly while a stream is held. An instance subscribes when one of the session's GET streams attaches to it, wherever the session was created, and unsubscribes when its last local stream for that session goes away. Because messages sent while no stream is open are already in history, the client gets them on reconnect with Last-Event-ID. Changes for one session are serialized (SessionTopicSubscriptions), so a quick close-and-reconnect can't have a late UNSUBSCRIBE drop the new subscription.
3. No gap on reconnect. A GET subscribes before it reads history for the replay. Live events that arrive during the replay are queued, then flushed without the ids the replay already sent. RedisMessageBroker.subscribe now waits for the SUBSCRIBE ack.
4. DELETE deletes the session, ends local streams, and publishes a termination packet on the session topic. Whichever instance holds the stream ends it.
5. The sweep runs for every store. It closes streams whose session is gone. With nothing subscribed at initialize, Redis backend leaks a broker subscription for every expired legacy session #329's leak can't happen in the first place. MemorySessionStore.cleanup() still runs as before.
6. RedisMessageBroker.unsubscribe tracks its listeners and removes them with removeListener, so the Redis channel is actually released.
7. Resource unsubscribe handlers are not reference-counted across eras and never run on legacy session expiry #337: per-instance holder count per URI (src/resource-subscription-holders.ts). An instance records the sessions it ran the subscribe handler for, since any watcher lives there. When a session stops holding a URI without an explicit resources/unsubscribe, the unsubscribe handler runs once no other session on that instance holds the URI. That covers DELETE, a termination from another instance, and, through the sweep, expiry (including Redis TTL expiry) and DELETE/unsubscribe handled by another instance. An explicit resources/unsubscribe still calls the handler directly, because its return value is the response. In those calls there is no request, so the unsubscribe handler's context is now ResourceUnsubscribeContext, with request/reply optional. That is a type-level change for TypeScript users who read context.request in an unsubscribe handler.

Raw messages published straight to the session topic (for example by the internal, unexported TokenRefreshService) still work: an instance that holds a stream records and delivers them. If no stream is open, they are not stored.

The commits are split: #327/#328/#329 first, then #337. I kept them in one PR because #337 relies on the end-of-session path (DELETE plus the all-store sweep) that the first commit introduces.

Possible conflict with #325

#325 (atomic SSE event ids) changes SessionStore.addMessage and sendSSEToStreams in src/routes/mcp.ts. Here, every history write goes through one helper, appendToHistory in src/session-topic.ts, used both by sendSSEToStreams and by the sender path. Once #325 lands, that helper becomes return sessionStore.addMessage(sessionId, message). Expect a textual conflict in sendSSEToStreams and in the removed subscriber-side history write in createSession.

Tests

• test/legacy-session-subscriptions.test.ts (new, real listening servers + undici, memory and Redis):
  • reconnect after the first stream closes: a message sent in the gap is replayed with Last-Event-ID, and live delivery resumes
  • a stream on instance B gets messages sent from A and from B, with history holding exactly ids 1 and 2
  • DELETE on A ends B's stream and leaves no subscriber (PUBSUB NUMSUB)
  • expired sessions, closed streams and vanished session keys leave no Redis subscription
  • the serialization race in SessionTopicSubscriptions
  • RedisMessageBroker.unsubscribe releases the channel
• test/resource-subscription-lifecycle.test.ts (new):
  • DELETE (with and without SSE) and expiry call the unsubscribe handler
  • with two sessions on one URI, the handler runs only for the last
  • no double call after an explicit unsubscribe
  • Redis: TTL expiry, DELETE via another instance, and unsubscribe via another instance are each released on the instance that ran the subscribe handler
• Before the fix, 7/8 tests in the first file and 7/8 in the second failed. The remaining one in each is a regression guard.
• test/broadcast-error-handling.test.ts: the two "session store fails" cases now expect mcpSendToSession to return false, and still check that nothing is unhandled and the error is logged.
• npm run lint, npm run typecheck, and npm run test (tsd + 784 tests, with REDIS_DB=3) all pass.

Fixes #327
Fixes #328
Fixes #329
Fixes #337

🤖 Generated with Claude Code

https://claude.ai/code/session_01LmhsFjk8W96X2PvmShmDm9