[CF-DATAPLANE] Downstream subscriptions/listen: context-carried admission and sink registry

[gandhipratik203]

Parent: #5681

Depends on / coordinates with: #5679, #5680
Blocks: #5346, #5615
Related: #5255

Goal

Implement the downstream MCP 2026-07-28 subscriptions/listen entry point in the Rust dataplane using RMCP's built-in subscription machinery, while keeping subscription admission accurate for the authenticated user and selected virtual host.

Implementation direction

Use RMCP 3.0.1's subscriptions/listen machinery. Do not hand-roll stream handling, acknowledgment, or io.modelcontextprotocol/subscriptionId tagging.

Carry resolved request context into the McpService instance used for modern stateless requests:

• authenticated principal
• selected VirtualHostId
• selected VirtualHost
• UserConfig as needed

McpService::get_info() should return capabilities derived from that carried context. accepted_subscription_filter() should narrow the requested filter against those instance capabilities. RMCP's default server/discover can be used if tests confirm it delegates to the context-aware get_info().

listen(context) owns the downstream sink registry: register SubscriptionSinks for accepted/routable kinds and clean them up when the listen stream closes.

Factory fallback paths without carried context must fail closed or return minimal safe capabilities. Backend-live capability refinement belongs with the stateless backend discovery/cache work in #5679.

Scope

• Add task-local or equivalent request-scoped context after middleware resolves claims, user config, and virtual host.
• Read that context in the StreamableHttpService factory and build McpService with principal/vhost/capability state.
• Make McpService::get_info() return context-derived capabilities.
• Make accepted_subscription_filter() narrow requested filters against those capabilities.
• Implement McpService::listen().
• Register SubscriptionSinks keyed by principal + virtual host + subscription id + notification kind.
• Clean up sinks on cancellation/stream close.
• Add fallback behavior for factory calls without request context.
• Add tests for accurate server/discover, subscription acceptance, sink cleanup, multiple subscriptions, and RMCP legacy/modern method gating.

Acceptance criteria

• A modern client can open subscriptions/listen without the gateway hand-rolling JSON-RPC stream behavior.
• server/discover reports capabilities derived from the authenticated user's selected virtual host.
• accepted_subscription_filter() narrows requested filters to the context-derived subscription capabilities.
• listen(context) registers sinks only for accepted/routable notification kinds.
• Sink state is removed when the listen stream is cancelled or closed.
• Multiple downstream subscriptions remain isolated from one another.
• Factory calls without request context fail closed or expose only minimal safe capabilities.
• Legacy subscriptions/listen returns method-not-found and modern resources/subscribe / resources/unsubscribe return method-not-found via RMCP gating.

Non-goals

• Relaying notifications/tools/list_changed, notifications/prompts/list_changed, or notifications/resources/list_changed; tracked in [CF-DATAPLANE] Relay MCP list-changed notifications through gateway #5346.
• Relaying notifications/resources/updated through subscriptions/listen; tracked in [CF-DATAPLANE] Relay MCP 2026-07-28 resource update notifications through subscriptions/listen #5615.
• Upstream subscriptions/listen stream management.
• Control-plane, UI, or management API changes.