Add downstream subscriptions/listen admission

_context/wiki/architecture.md

@@ -84,8 +84,10 @@ Order is invariant: auth/config before backend selection; request plugins before
 | JWT decoders | `ContextForgeDataPlaneAppState` | Process |
 | User config | `RedisUserConfigStore` (LRU + Redis) | Request-path consumed; control-plane authored |
 | Request identity / VirtualHostId | Request extensions | One HTTP request |
+| Gateway request context | `virtual_host_config_layer` task-local | One HTTP request; copied into the RMCP service factory for context-aware local methods |
 | Downstream session id | RMCP + `SessionId` extension | MCP session |
 | Backend RMCP services | `BackendTransports` map | Local process, per principal/backend/session |
+| Downstream subscription sinks | `DownstreamSubscriptionRegistry` | Listen-stream lifetime, keyed by principal, virtual host, subscription id, registration id, and notification kind |
 | Local user session mapping | `LocalUserSessionStore` | Local LRU, 50k entries, 1 hour |
 | Plugin manager | `CpexRuntimeRegistry` | Process, reloadable |
 
@@ -107,6 +109,7 @@ In multi-runtime mode, the first thread initializes the optional CPEX plugin run
 | State | Lock | Contention profile |
 | --- | --- | --- |
 | `BackendTransports` map | `Arc<tokio::sync::Mutex<HashMap<...>>>` | Locked briefly on initialize insert, per-call borrow, and cleanup. Borrowing clones `Arc<RunningService>` handles so the lock is not held across backend calls. |
+| `DownstreamSubscriptionRegistry` map | `Arc<std::sync::Mutex<HashMap<...>>>` | Locked only for short insert/remove operations. Cleanup runs from `Drop`, so it cannot await. No registry lock is held across stream waits or backend I/O. |
 | Subscription set | `Arc<tokio::sync::Mutex<HashSet<String>>>` | Local `subscribe`/`unsubscribe` only. |
 | User config LRU cache | `Arc<tokio::sync::Mutex<LruCache>>` inside `RedisUserConfigStore` | One lock per config lookup on the hot path; misses add a Redis round trip. |
 | User session LRU cache | Same pattern in `LocalUserSessionStore` | Initialize and delete paths. |
@@ -128,6 +131,7 @@ The binary sets `tikv_jemallocator` as the global allocator. jemalloc holds up b
 - List methods fan out to all connected backends concurrently and merge.
 - Targeted calls resolve exactly one backend service handle.
 - `call_tool` watches the downstream cancellation token and forwards a cancel to the backend if the client gives up first; backend progress notifications are forwarded downstream while the call is in flight.
+- `subscriptions/listen` registers downstream sinks, parks on RMCP cancellation, and removes those sinks when the listen stream closes.
 
 ## Startup And Response Flow
 

_context/wiki/routing.md

@@ -60,6 +60,57 @@ This is **local process state only**. Implications:
 - Gateway restart → all sessions lost → clients must re-run `initialize`.
 - Multi-runtime mode (`--single-runtime false`): each runtime thread has its own `BackendTransports` with no cross-thread affinity.
 
+## Modern Discovery And Subscription Admission
+
+Modern downstream clients use `server/discover` with MCP `2026-07-28`.
+`McpService::supported_protocol_versions()` advertises only `2026-07-28`, so
+requests that declare older protocol versions fail protocol-version validation.
+Full removal of RMCP legacy session behavior belongs to the stateless routing
+migration tracked separately.
+
+For normal requests, `virtual_host_config_layer` carries the authenticated
+principal, selected virtual host id, and selected `VirtualHost` into the RMCP
+service factory. The per-request `McpService` then uses that context for local
+modern methods:
+
+```text
+MCP client
+  -> /servers/{virtual_host_id}/mcp
+  -> auth + user config + virtual host check
+  -> RMCP service factory receives principal + virtual host
+  -> server/discover and subscriptions/listen use that context
+```
+
+`server/discover` reports capabilities derived from the selected virtual host.
+Today this is vhost-accurate, not backend-live-accurate: a non-empty virtual
+host advertises list-change and resource subscription support, while backend
+capability-cache refinement belongs to the stateless routing work.
+
+`subscriptions/listen` uses RMCP's subscription machinery. The gateway narrows
+the requested filter to supported notification kinds and routable resource
+subscription URIs, registers the accepted `SubscriptionSink`s in
+`DownstreamSubscriptionRegistry`, and removes them when the listen stream is
+cancelled or closed.
+
+```text
+MCP client
+   |
+   | subscriptions/listen
+   v
+Gateway
+   |
+   | narrow filter against selected virtual host
+   | register accepted sinks
+   | wait for listen cancellation
+   | remove sinks on close
+   v
+DownstreamSubscriptionRegistry
+```
+
+Notification delivery is intentionally separate follow-up work:
+`*/list_changed` relay, `resources/updated` relay over `subscriptions/listen`,
+and upstream `subscriptions/listen` management.
+
 
 ```mermaid
 sequenceDiagram
@@ -131,4 +182,6 @@ If RMCP rejects the delete, local state is untouched.
 | `get_prompt` | Targeted | Single-backend: name unchanged. Multi-backend: strips prefix. |
 | `complete` | Targeted | Routes on prompt name or resource URI inside `ref`. |
 | `ping` | Local | Returns success; no backend fanout. |
+| `server/discover` | Local | Reports `2026-07-28` support and capabilities derived from the authenticated user's selected virtual host. |
+| `subscriptions/listen` | Local | Narrows the requested subscription filter, registers downstream sinks, and cleans them up when the listen stream closes. Backend notification delivery is follow-up work. |
 | `DELETE` | Session | RMCP handles first; on success `session_id_layer` removes local session + backend transports. |

crates/contextforge-data-plane-lib/src/gateway/downstream_subscriptions.rs

@@ -0,0 +1,207 @@
+use std::{
+    collections::HashMap,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicU64, Ordering},
+    },
+};
+
+use rmcp::{
+    model::{RequestId, SubscriptionFilter},
+    service::SubscriptionSink,
+};
+
+use crate::layers::request_context::GatewayRequestContext;
+
+#[derive(Clone, Default)]
+pub(crate) struct DownstreamSubscriptionRegistry {
+    inner: Arc<Mutex<HashMap<DownstreamSubscriptionKey, SubscriptionSink>>>,
+    next_registration_id: Arc<AtomicU64>,
+}
+
+impl DownstreamSubscriptionRegistry {
+    pub(crate) fn register(
+        &self,
+        context: &GatewayRequestContext,
+        filter: &SubscriptionFilter,
+        sink: &SubscriptionSink,
+    ) -> DownstreamSubscriptionGuard {
+        let registration_id = self.next_registration_id.fetch_add(1, Ordering::Relaxed);
+        let keys = subscription_keys(context, filter, sink.id(), registration_id);
+        let mut subscriptions = self.inner.lock().expect("downstream subscription registry lock poisoned");
+        for key in &keys {
+            subscriptions.insert(key.clone(), sink.clone());
+        }
+        DownstreamSubscriptionGuard { registry: self.clone(), keys }
+    }
+
+    fn remove_all(&self, keys: &[DownstreamSubscriptionKey]) {
+        let mut subscriptions = self.inner.lock().expect("downstream subscription registry lock poisoned");
+        for key in keys {
+            subscriptions.remove(key);
+        }
+    }
+}
+
+pub(crate) struct DownstreamSubscriptionGuard {
+    registry: DownstreamSubscriptionRegistry,
+    keys: Vec<DownstreamSubscriptionKey>,
+}
+
+impl Drop for DownstreamSubscriptionGuard {
+    fn drop(&mut self) {
+        self.registry.remove_all(&self.keys);
+    }
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub(crate) struct DownstreamSubscriptionKey {
+    principal: String,
+    virtual_host_id: String,
+    subscription_id: RequestId,
+    registration_id: u64,
+    notification: DownstreamSubscriptionNotification,
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub(crate) enum DownstreamSubscriptionNotification {
+    ToolsListChanged,
+    PromptsListChanged,
+    ResourcesListChanged,
+    ResourceUpdated { uri: String },
+}
+
+pub(super) fn subscription_keys(
+    context: &GatewayRequestContext,
+    filter: &SubscriptionFilter,
+    subscription_id: &RequestId,
+    registration_id: u64,
+) -> Vec<DownstreamSubscriptionKey> {
+    let mut keys = Vec::new();
+    if filter.tools_list_changed == Some(true) {
+        keys.push(subscription_key(
+            context,
+            subscription_id,
+            registration_id,
+            DownstreamSubscriptionNotification::ToolsListChanged,
+        ));
+    }
+    if filter.prompts_list_changed == Some(true) {
+        keys.push(subscription_key(
+            context,
+            subscription_id,
+            registration_id,
+            DownstreamSubscriptionNotification::PromptsListChanged,
+        ));
+    }
+    if filter.resources_list_changed == Some(true) {
+        keys.push(subscription_key(
+            context,
+            subscription_id,
+            registration_id,
+            DownstreamSubscriptionNotification::ResourcesListChanged,
+        ));
+    }
+    if let Some(uris) = &filter.resource_subscriptions {
+        keys.extend(uris.iter().map(|uri| {
+            subscription_key(
+                context,
+                subscription_id,
+                registration_id,
+                DownstreamSubscriptionNotification::ResourceUpdated { uri: uri.clone() },
+            )
+        }));
+    }
+    keys
+}
+
+fn subscription_key(
+    context: &GatewayRequestContext,
+    subscription_id: &RequestId,
+    registration_id: u64,
+    notification: DownstreamSubscriptionNotification,
+) -> DownstreamSubscriptionKey {
+    DownstreamSubscriptionKey {
+        principal: context.principal().to_owned(),
+        virtual_host_id: context.virtual_host_id().to_owned(),
+        subscription_id: subscription_id.clone(),
+        registration_id,
+        notification,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use contextforge_data_plane_apis::user_store::{BackendMCPGateway, VirtualHost};
+    use rmcp::model::RequestId;
+
+    use super::*;
+
+    #[test]
+    fn keys_include_subscription_id_and_notification_kind() {
+        let gateway_context = GatewayRequestContext::new(&test_claims(), &test_virtual_host_id(), &test_virtual_host());
+        let filter = SubscriptionFilter::builder().tools_list_changed().resource_subscription("memo://known").build();
+
+        let keys = subscription_keys(&gateway_context, &filter, &RequestId::Number(7), 9);
+
+        assert_eq!(2, keys.len());
+        assert!(keys.iter().all(|key| key.subscription_id == RequestId::Number(7)));
+        assert!(keys.iter().all(|key| key.registration_id == 9));
+    }
+
+    #[test]
+    fn registration_id_is_part_of_key_identity() {
+        let gateway_context = GatewayRequestContext::new(&test_claims(), &test_virtual_host_id(), &test_virtual_host());
+        let filter = SubscriptionFilter::builder().tools_list_changed().build();
+
+        let first = subscription_keys(&gateway_context, &filter, &RequestId::Number(7), 0);
+        let second = subscription_keys(&gateway_context, &filter, &RequestId::Number(7), 1);
+
+        assert_ne!(first, second);
+    }
+
+    fn test_virtual_host() -> VirtualHost {
+        VirtualHost {
+            backends: HashMap::from([(
+                "backend-one".to_owned(),
+                BackendMCPGateway {
+                    name: "backend-one".to_owned(),
+                    url: "http://127.0.0.1:9999/mcp".parse().expect("valid URL"),
+                    passthrough_headers: Vec::new(),
+                    add_headers: HashMap::new(),
+                    remove_headers: Vec::new(),
+                    allowed_tool_names: Vec::new(),
+                    tool_name_aliases: HashMap::new(),
+                    allowed_resource_names: Vec::new(),
+                    allowed_prompt_names: Vec::new(),
+                },
+            )]),
+        }
+    }
+
+    fn test_claims() -> crate::common::ContextForgeClaims {
+        crate::common::ContextForgeClaims {
+            sub: "test-principal".to_owned(),
+            jti: "test-jti".to_owned(),
+            token_use: None,
+            iat: None,
+            iss: "test-issuer".to_owned(),
+            aud: "test-audience".to_owned(),
+            exp: 1,
+            teams: None,
+            user: crate::common::User::builder()
+                .email("test@example.com".to_owned())
+                .full_name(None)
+                .is_admin(false)
+                .auth_provider("test".to_owned())
+                .build(),
+            scopes: None,
+        }
+    }
+
+    fn test_virtual_host_id() -> crate::layers::virtual_host_id::VirtualHostId {
+        crate::layers::virtual_host_id::VirtualHostId::new("test-vhost".to_owned())
+    }
+}