feat(vllm): add Prime RL protocol parity

lib/backend-common/src/lib.rs

@@ -42,6 +42,7 @@ pub use engine::{
 };
 pub use error::{BackendError, DynamoError, ErrorType};
 pub use metrics::{ComponentGauges, EngineMetrics, LifecycleGauges};
+pub use rl::RlWorkerMetadata;
 pub use run::{run, run_raw};
 pub use snapshot_publisher::SnapshotPublisher;
 pub use worker::{RuntimeConfig, Worker, WorkerConfig};

lib/backend-common/src/rl.rs

@@ -1,7 +1,7 @@
 // SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::sync::Arc;
+use std::{num::NonZeroU32, sync::Arc};
 
 use async_trait::async_trait;
 use dynamo_runtime::component::{Endpoint, StartedEndpoint};
@@ -24,6 +24,48 @@ pub(crate) struct RlServeEndpoint {
 pub(crate) struct RlEndpointConfig {
     endpoint_name: String,
     system_url: String,
+    metadata: Option<RlWorkerMetadata>,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct RlWorkerMetadata {
+    world_size: NonZeroU32,
+    weight_transfer_backend: Option<String>,
+    admin_base_url: Option<String>,
+}
+
+impl RlWorkerMetadata {
+    pub fn new(
+        world_size: u32,
+        weight_transfer_backend: Option<String>,
+        admin_base_url: Option<String>,
+    ) -> anyhow::Result<Self> {
+        let world_size = NonZeroU32::new(world_size)
+            .ok_or_else(|| anyhow::anyhow!("RL worker world size must be positive"))?;
+        let weight_transfer_backend = normalized_optional(
+            weight_transfer_backend,
+            "RL weight-transfer backend must not be blank",
+        )?;
+        let admin_base_url =
+            normalized_optional(admin_base_url, "RL admin base URL must not be blank")?;
+        Ok(Self {
+            world_size,
+            weight_transfer_backend,
+            admin_base_url,
+        })
+    }
+}
+
+fn normalized_optional(value: Option<String>, error: &str) -> anyhow::Result<Option<String>> {
+    value
+        .map(|value| {
+            let value = value.trim();
+            if value.is_empty() {
+                anyhow::bail!(error.to_string());
+            }
+            Ok(value.to_string())
+        })
+        .transpose()
 }
 
 impl RlServeEndpoint {
@@ -32,7 +74,10 @@ impl RlServeEndpoint {
     }
 }
 
-pub(crate) fn prepare_endpoint(primary: &Endpoint) -> anyhow::Result<RlEndpointConfig> {
+pub(crate) fn prepare_endpoint(
+    primary: &Endpoint,
+    metadata: Option<RlWorkerMetadata>,
+) -> anyhow::Result<RlEndpointConfig> {
     let endpoint_name = resolve_endpoint_name(&primary.id().name)?;
     let system_url = self_host_base_url(primary.drt()).ok_or_else(|| {
         anyhow::anyhow!(
@@ -42,6 +87,7 @@ pub(crate) fn prepare_endpoint(primary: &Endpoint) -> anyhow::Result<RlEndpointC
     Ok(RlEndpointConfig {
         endpoint_name,
         system_url,
+        metadata,
     })
 }
 
@@ -53,6 +99,7 @@ pub(crate) async fn serve_endpoint(
     let handler = Arc::new(RlRouteHandler {
         routes: primary.drt().engine_routes().clone(),
         system_url: config.system_url,
+        metadata: config.metadata,
     });
     let ingress = Ingress::for_engine(handler)?;
     let started = endpoint
@@ -100,6 +147,7 @@ fn validate_endpoint_name(endpoint_name: &str, primary_name: &str) -> anyhow::Re
 struct RlRouteHandler {
     routes: EngineRouteRegistry,
     system_url: String,
+    metadata: Option<RlWorkerMetadata>,
 }
 
 impl RlRouteHandler {
@@ -122,11 +170,21 @@ impl RlRouteHandler {
         let mut routes = self.routes.routes().into_iter().collect::<Vec<_>>();
         routes.sort();
         routes.dedup();
-        json!({
+        let mut response = json!({
             "status": "ok",
             "routes": routes,
             "system_url": self.system_url,
-        })
+        });
+        if let Some(metadata) = &self.metadata {
+            response["world_size"] = json!(metadata.world_size.get());
+            if let Some(backend) = &metadata.weight_transfer_backend {
+                response["weight_transfer_backend"] = json!(backend);
+            }
+            if let Some(url) = &metadata.admin_base_url {
+                response["admin_base_url"] = json!(url);
+            }
+        }
+        response
     }
 }
 
@@ -156,6 +214,14 @@ mod tests {
         let handler = RlRouteHandler {
             routes,
             system_url: "http://worker:8080".to_string(),
+            metadata: Some(
+                RlWorkerMetadata::new(
+                    4,
+                    Some(" nccl ".to_string()),
+                    Some(" http://worker:8120 ".to_string()),
+                )
+                .expect("valid metadata"),
+            ),
         };
 
         assert_eq!(
@@ -164,11 +230,21 @@ mod tests {
                 "status": "ok",
                 "routes": ["control/pause_generation"],
                 "system_url": "http://worker:8080",
+                "admin_base_url": "http://worker:8120",
+                "world_size": 4,
+                "weight_transfer_backend": "nccl",
             })
         );
         assert_eq!(
             handler.dispatch(&json!({"method": "control/pause_generation"}))["status"],
             "error"
         );
     }
+
+    #[test]
+    fn rl_worker_metadata_rejects_invalid_values() {
+        assert!(RlWorkerMetadata::new(0, None, None).is_err());
+        assert!(RlWorkerMetadata::new(1, Some("   ".to_string()), None).is_err());
+        assert!(RlWorkerMetadata::new(1, None, Some("   ".to_string())).is_err());
+    }
 }

lib/backend-common/src/worker.rs

@@ -193,6 +193,8 @@ pub struct WorkerConfig {
     pub route_to_encoder: bool,
     /// Publish the worker's engine routes through an auxiliary RL discovery endpoint.
     pub enable_rl: bool,
+    /// Optional RL topology and weight-transfer metadata published by the worker.
+    pub rl_metadata: Option<crate::RlWorkerMetadata>,
     /// Optional frontend media decoding and fetch policy advertised on the
     /// model deployment card.
     pub media_decoder: Option<MediaDecoder>,
@@ -238,6 +240,7 @@ impl Default for WorkerConfig {
             runtime: RuntimeConfig::default(),
             route_to_encoder: false,
             enable_rl: false,
+            rl_metadata: None,
             media_decoder: None,
             media_fetcher: None,
             default_thinking_mode: None,
@@ -926,12 +929,16 @@ impl Worker {
         let model_type = resolve_model_type(&self.config)?;
         let (worker_type, needs) = resolve_worker_type_and_needs(&self.config);
         let rl_config = if self.config.enable_rl {
-            Some(crate::rl::prepare_endpoint(&endpoint).map_err(|error| {
-                err(
-                    ErrorType::Backend(BackendError::InvalidArgument),
-                    format!("RL endpoint configuration: {error}"),
-                )
-            })?)
+            Some(
+                crate::rl::prepare_endpoint(&endpoint, self.config.rl_metadata.clone()).map_err(
+                    |error| {
+                        err(
+                            ErrorType::Backend(BackendError::InvalidArgument),
+                            format!("RL endpoint configuration: {error}"),
+                        )
+                    },
+                )?,
+            )
         } else {
             None
         };

lib/bindings/python/rust/backend.rs

@@ -469,6 +469,7 @@ impl WorkerConfig {
                 // Python vLLM owns and serves its existing `.rl` endpoint.
                 // The shared Rust endpoint is opt-in for Rust sidecars only.
                 enable_rl: false,
+                rl_metadata: None,
                 media_decoder: media_decoder.map(|decoder| decoder.inner),
                 media_fetcher: media_fetcher.map(|fetcher| fetcher.inner),
             },

lib/kv-router/src/protocols.rs

@@ -79,6 +79,21 @@ pub fn pad_value_for_mm_hash(mm_hash: u64) -> u32 {
     (MM_PAD_SHIFT_VALUE + (mm_hash & MM_PAD_HASH_MASK)) as u32
 }
 
+/// Map a non-empty multimodal identifier to Dynamo's routing hash.
+pub fn hash_mm_identifier(identifier: &str) -> Option<u64> {
+    if identifier.is_empty() {
+        return None;
+    }
+    if identifier.len() == 64
+        && identifier
+            .chars()
+            .all(|character| character.is_ascii_hexdigit())
+    {
+        return u64::from_str_radix(&identifier[..16], 16).ok();
+    }
+    Some(xxh3::xxh3_64(identifier.as_bytes()))
+}
+
 /// Compute the hash for a sequence of tokens, optionally including multimodal metadata,
 /// LoRA adapter identity, and cache namespace.
 ///
@@ -1865,6 +1880,18 @@ mod tests {
         );
     }
 
+    #[test]
+    fn mm_identifier_hash_preserves_vllm_and_opaque_identifiers() {
+        let canonical = "0123456789abcdef".repeat(4);
+        assert_eq!(hash_mm_identifier(&canonical), Some(0x0123_4567_89ab_cdef));
+        let opaque = "opaque-renderer-image-0";
+        assert_eq!(
+            hash_mm_identifier(opaque),
+            Some(xxh3::xxh3_64(opaque.as_bytes()))
+        );
+        assert_eq!(hash_mm_identifier(""), None);
+    }
+
     #[test]
     fn test_router_event_new() {
         let worker_id = 0;

lib/llm/src/discovery.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 mod model;
-pub use model::Model;
+pub use model::{GenerateEngineSelection, Model};
 
 pub mod kv_source_membership;
 pub use kv_source_membership::{

lib/llm/src/discovery/model.rs

@@ -17,6 +17,7 @@ use super::worker_monitor::LoadThresholdConfig;
 use super::worker_set::WorkerSet;
 use crate::protocols::openai::ParsingOptions;
 
+use crate::local_model::runtime_config::VLLM_EXACT_MM_ROUTING_CAPABILITY;
 use crate::types::{
     RealtimeBidirectionalEngine,
     generic::tensor::TensorStreamingEngine,
@@ -30,6 +31,13 @@ use crate::types::{
     },
 };
 
+#[derive(Clone)]
+pub struct GenerateEngineSelection {
+    pub engine: GenerateStreamingEngine,
+    pub kv_cache_block_size: u32,
+    pub supports_exact_mm_routing: bool,
+}
+
 /// Emit a one-time deprecation warning when serving-readiness falls back to
 /// the legacy path because a namespace still contains a legacy card (a
 /// worker with no declared `worker_type`). Logged once per process to avoid
@@ -671,6 +679,29 @@ impl Model {
         .ok_or_else(|| self.engine_error(self.has_generate_engine_for_capability(capability)))
     }
 
+    pub fn get_generate_engine_selection_for_capability(
+        &self,
+        capability: &str,
+    ) -> Result<GenerateEngineSelection, ModelManagerError> {
+        self.select_worker_set_with(|worker_set| {
+            worker_set
+                .supports_runtime_capability(capability)
+                .then(|| {
+                    worker_set
+                        .generate_engine
+                        .clone()
+                        .map(|engine| GenerateEngineSelection {
+                            engine,
+                            kv_cache_block_size: worker_set.card().kv_cache_block_size,
+                            supports_exact_mm_routing: worker_set
+                                .supports_runtime_capability(VLLM_EXACT_MM_ROUTING_CAPABILITY),
+                        })
+                })
+                .flatten()
+        })
+        .ok_or_else(|| self.engine_error(self.has_generate_engine_for_capability(capability)))
+    }
+
     // -- Combined engine + parsing options (atomically from one WorkerSet) --
 
     pub fn get_chat_engine_with_parsing(

lib/llm/src/discovery/model_manager.rs

@@ -20,7 +20,7 @@ use dynamo_kv_router::{
 
 use super::worker_monitor::LoadThresholdConfig;
 use super::{
-    KvSourceMembershipWatch, Model, RuntimeConfigWatch, WorkerSet,
+    GenerateEngineSelection, KvSourceMembershipWatch, Model, RuntimeConfigWatch, WorkerSet,
     kv_source_watch::KvSourceMembershipCoordinator, runtime_config_watch,
 };
 
@@ -1342,6 +1342,19 @@ impl ModelManager {
             .get_generate_engine_for_capability(capability)
     }
 
+    pub fn get_generate_engine_selection_for_capability(
+        &self,
+        model: &str,
+        capability: &str,
+    ) -> Result<GenerateEngineSelection, ModelManagerError> {
+        self.catalog
+            .load()
+            .models
+            .get(model)
+            .ok_or_else(|| ModelManagerError::ModelNotFound(model.to_string()))?
+            .get_generate_engine_selection_for_capability(capability)
+    }
+
     // -- Combined engine + parsing options (atomically from one WorkerSet) --
 
     pub fn get_chat_completions_engine_with_parsing(