feat(vllm): route RL controls through sidecar

.github/workflows/copyright-check.ps1

@@ -84,7 +84,7 @@ $global:copyright_results = @{
 
 $ignored_files = @('.clang-format', '.gitattributes', '.gitignore', '.gitkeep', '.patch', 'Cargo.lock', 'LICENSE', 'uv.lock', 'rust-toolchain.toml', 'codespell.txt', 'exclusions.txt')
 write-debug "<copyright-check> ignored_files = ['$($ignored_files -join "','")']."
-$ignored_paths = @('.github', '.mypy_cache', '.pytest_cache', 'lib/llm/tests/data/sample-models', 'lib/llm/tests/data/deepseek-v3.2', 'lib/llm/tests/data/deepseek-v4', 'container/compliance/spdx_licenses', 'lib/sidecar/vllm/proto/vllm_grpc.proto')
+$ignored_paths = @('.github', '.mypy_cache', '.pytest_cache', 'lib/llm/tests/data/sample-models', 'lib/llm/tests/data/deepseek-v3.2', 'lib/llm/tests/data/deepseek-v4', 'container/compliance/spdx_licenses', 'lib/sidecar/vllm/proto/control.proto', 'lib/sidecar/vllm/proto/inference.proto')
 write-debug "<copyright-check> ignored_paths = ['$($ignored_paths -join "','")']."
 $ignored_types = @('.bat', '.gif', '.ico', '.ipynb', '.jpg', '.jpeg', '.patch', '.png', '.pyc', '.pyi', '.rst', '.zip', '.md', '.json')
 write-debug "<copyright-check> ignored_types = ['$($ignored_types -join "', '")']."

.pre-commit-config.yaml

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 default_install_hook_types: [pre-commit, commit-msg]
-exclude: ^(src/grpc_generated|.*\.patch$|.*/connect/.*\.py|components/src/dynamo/planner/plugins/proto/v1/plugin_pb2(_grpc)?\.pyi?$|lib/sidecar/vllm/proto/vllm_grpc\.proto$|lib/sidecar/trtllm/proto/trtllm_service\.proto$)
+exclude: ^(src/grpc_generated|.*\.patch$|.*/connect/.*\.py|components/src/dynamo/planner/plugins/proto/v1/plugin_pb2(_grpc)?\.pyi?$|lib/sidecar/vllm/proto/(control|inference)\.proto$|lib/sidecar/trtllm/proto/trtllm_service\.proto$)
 repos:
 - repo: https://github.com/timothycrosley/isort
   rev: 5.12.0

lib/backend-common/examples/mocker/src/engine.rs

@@ -275,6 +275,7 @@ impl MockerBackend {
             custom_jinja_template: args.common.custom_jinja_template,
             disaggregation_mode,
             route_to_encoder: args.common.route_to_encoder,
+            enable_rl: args.common.enable_rl,
             model_name: args.model_path,
             served_model_name: Some(args.model_name),
             tool_call_parser,

lib/backend-common/src/args.rs

@@ -90,4 +90,8 @@ pub struct CommonArgs {
     /// shim does not read this env var.
     #[arg(long, default_value_t = false, env = "DYN_ROUTE_TO_ENCODER")]
     pub route_to_encoder: bool,
+
+    /// Publish this worker's engine control/update routes on the RL request-plane endpoint.
+    #[arg(long, default_value_t = false, env = "DYN_ENABLE_RL")]
+    pub enable_rl: bool,
 }

lib/backend-common/src/lib.rs

@@ -19,6 +19,7 @@ pub mod engine;
 pub mod error;
 pub mod metrics;
 mod publisher;
+mod rl;
 pub mod run;
 pub mod snapshot_publisher;
 pub mod telemetry;

lib/backend-common/src/rl.rs

@@ -0,0 +1,152 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use dynamo_runtime::component::{Endpoint, StartedEndpoint};
+use dynamo_runtime::engine_routes::EngineRouteRegistry;
+use dynamo_runtime::pipeline::network::Ingress;
+use dynamo_runtime::pipeline::{
+    AsyncEngine, AsyncEngineContextProvider, ManyOut, ResponseStream, SingleIn,
+};
+use dynamo_runtime::protocols::annotated::Annotated;
+use dynamo_runtime::traits::DistributedRuntimeProvider;
+use futures::stream;
+use serde_json::{Value, json};
+
+const DEFAULT_RL_ENDPOINT: &str = "rl";
+
+pub(crate) struct RlServeEndpoint {
+    started: StartedEndpoint,
+}
+
+impl RlServeEndpoint {
+    pub(crate) async fn shutdown(self) -> anyhow::Result<()> {
+        self.started.shutdown().await
+    }
+}
+
+pub(crate) async fn serve_endpoint(primary: &Endpoint) -> anyhow::Result<RlServeEndpoint> {
+    let endpoint_name = resolve_endpoint_name(&primary.id().name)?;
+    let endpoint = primary.component().endpoint(endpoint_name);
+    let system_url = self_host_base_url(primary.drt());
+    let handler = Arc::new(RlRouteHandler {
+        routes: primary.drt().engine_routes().clone(),
+        system_url,
+    });
+    let ingress = Ingress::for_engine(handler)?;
+    let started = endpoint
+        .endpoint_builder()
+        .handler(ingress)
+        .graceful_shutdown(true)
+        .start_with_registration()
+        .await?;
+    Ok(RlServeEndpoint { started })
+}
+
+fn self_host_base_url(drt: &dynamo_runtime::DistributedRuntime) -> Option<String> {
+    let info = drt.system_status_server_info()?;
+    let configured = dynamo_runtime::RuntimeConfig::from_settings()
+        .unwrap_or_default()
+        .system_host;
+    let host = match configured.as_str() {
+        "0.0.0.0" | "::" | "[::]" => dynamo_runtime::utils::local_ip_for_advertise(),
+        _ => configured,
+    };
+    Some(format!("http://{host}:{}", info.port()))
+}
+
+fn resolve_endpoint_name(primary_name: &str) -> anyhow::Result<String> {
+    let endpoint_name =
+        std::env::var("DYN_RL_ENDPOINT").unwrap_or_else(|_| DEFAULT_RL_ENDPOINT.into());
+    validate_endpoint_name(endpoint_name.trim(), primary_name)
+}
+
+fn validate_endpoint_name(endpoint_name: &str, primary_name: &str) -> anyhow::Result<String> {
+    if endpoint_name.is_empty() {
+        anyhow::bail!("DYN_RL_ENDPOINT must not be empty");
+    }
+    if endpoint_name == primary_name {
+        anyhow::bail!("DYN_RL_ENDPOINT `{endpoint_name}` collides with the serving endpoint");
+    }
+    if !endpoint_name
+        .chars()
+        .all(|character| character.is_ascii_alphanumeric() || matches!(character, '-' | '_'))
+    {
+        anyhow::bail!("DYN_RL_ENDPOINT must contain only letters, digits, '-' or '_'");
+    }
+    Ok(endpoint_name.to_string())
+}
+
+struct RlRouteHandler {
+    routes: EngineRouteRegistry,
+    system_url: Option<String>,
+}
+
+impl RlRouteHandler {
+    fn dispatch(&self, request: &Value) -> Value {
+        let Some(method) = request
+            .as_object()
+            .and_then(|request| request.get("method"))
+            .and_then(Value::as_str)
+        else {
+            return json!({"status": "error", "message": "rl_dispatch: missing 'method' (str)"});
+        };
+        if method != "routes" {
+            return json!({
+                "status": "error",
+                "method": method,
+                "message": "rl request-plane endpoint only supports method='routes'",
+            });
+        }
+
+        let mut routes = self.routes.routes().into_iter().collect::<Vec<_>>();
+        routes.sort();
+        routes.dedup();
+        json!({
+            "status": "ok",
+            "routes": routes,
+            "system_url": self.system_url,
+        })
+    }
+}
+
+#[async_trait]
+impl AsyncEngine<SingleIn<Value>, ManyOut<Annotated<Value>>, anyhow::Error> for RlRouteHandler {
+    async fn generate(&self, input: SingleIn<Value>) -> anyhow::Result<ManyOut<Annotated<Value>>> {
+        let (request, context) = input.into_parts();
+        let response = self.dispatch(&request);
+        Ok(ResponseStream::new(
+            Box::pin(stream::once(async move { Annotated::from_data(response) })),
+            context.context(),
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn routes_request_describes_the_worker_engine_surface() {
+        let routes = EngineRouteRegistry::new();
+        routes.register(
+            "control/pause_generation",
+            Arc::new(|_| Box::pin(async { Ok(json!({"status": "ok"})) })),
+        );
+        let handler = RlRouteHandler {
+            routes,
+            system_url: Some("http://worker:8080".to_string()),
+        };
+
+        assert_eq!(
+            handler.dispatch(&json!({"method": "routes"})),
+            json!({
+                "status": "ok",
+                "routes": ["control/pause_generation"],
+                "system_url": "http://worker:8080",
+            })
+        );
+    }
+}

lib/backend-common/src/worker.rs

@@ -187,6 +187,8 @@ pub struct WorkerConfig {
     /// roles -- setting it on `Decode` or `Encode` is rejected at
     /// `Worker::run` validation time with `BackendError::InvalidArgument`.
     pub route_to_encoder: bool,
+    /// Publish the worker's engine routes through an auxiliary RL discovery endpoint.
+    pub enable_rl: bool,
     /// Optional frontend media decoding and fetch policy advertised on the
     /// model deployment card.
     pub media_decoder: Option<MediaDecoder>,
@@ -231,6 +233,7 @@ impl Default for WorkerConfig {
             structural_tag_schema: StructuralTagSchemaMode::Auto,
             runtime: RuntimeConfig::default(),
             route_to_encoder: false,
+            enable_rl: false,
             media_decoder: None,
             media_fetcher: None,
             default_thinking_mode: None,
@@ -982,7 +985,22 @@ impl Worker {
         let serve_fut = builder.start();
         tokio::pin!(serve_fut);
 
-        tokio::select! {
+        let rl_endpoint = if self.config.enable_rl {
+            match crate::rl::serve_endpoint(&endpoint).await {
+                Ok(endpoint) => Some(endpoint),
+                Err(error) => {
+                    self.orchestrator_steps(&endpoint).await;
+                    return Err(err(
+                        ErrorType::Backend(BackendError::Unknown),
+                        format!("RL endpoint setup: {error}"),
+                    ));
+                }
+            }
+        } else {
+            None
+        };
+
+        let serve_result = tokio::select! {
             biased;
             result = &mut serve_fut => {
                 match result {
@@ -993,23 +1011,31 @@ impl Worker {
                         tracing::info!(
                             "Endpoint completed gracefully; running shutdown orchestration"
                         );
+                        Ok(())
                     }
                     // Serve errored; cleanup_once in run() is the safety net.
                     Err(e) => {
-                        return Err(err(
+                        Err(err(
                             ErrorType::Backend(BackendError::Unknown),
                             format!("serve: {e}"),
-                        ));
+                        ))
                     }
                 }
             }
             _ = shutdown.cancelled() => {
                 tracing::info!("Received shutdown signal; running graceful orchestration");
+                Ok(())
             }
+        };
+
+        if let Some(rl_endpoint) = rl_endpoint
+            && let Err(error) = rl_endpoint.shutdown().await
+        {
+            tracing::warn!(%error, "RL discovery endpoint shutdown failed");
         }
 
         self.orchestrator_steps(&endpoint).await;
-        Ok(())
+        serve_result
     }
 
     /// Engine-facing shutdown sequence: grace period sleep → drain loop on
@@ -1287,8 +1313,12 @@ fn engine_control_policy(control: &str) -> EngineControlPolicy {
         // Pause controls make the engine unsafe for new requests, so remove
         // the endpoint before they mutate engine state. Resume controls make
         // the engine serving-safe again, so advertise it only after success.
-        "sleep" | "release_memory_occupation" => EngineControlPolicy::UnregisterBefore,
-        "wake_up" | "resume_memory_occupation" => EngineControlPolicy::RegisterAfter,
+        "pause_generation" | "sleep" | "release_memory_occupation" => {
+            EngineControlPolicy::UnregisterBefore
+        }
+        "resume_generation" | "wake_up" | "resume_memory_occupation" => {
+            EngineControlPolicy::RegisterAfter
+        }
         _ => EngineControlPolicy::Direct,
     }
 }
@@ -1825,6 +1855,10 @@ mod tests {
             engine_control_policy("sleep"),
             EngineControlPolicy::UnregisterBefore
         );
+        assert_eq!(
+            engine_control_policy("pause_generation"),
+            EngineControlPolicy::UnregisterBefore
+        );
         assert_eq!(
             engine_control_policy("release_memory_occupation"),
             EngineControlPolicy::UnregisterBefore
@@ -1833,6 +1867,10 @@ mod tests {
             engine_control_policy("wake_up"),
             EngineControlPolicy::RegisterAfter
         );
+        assert_eq!(
+            engine_control_policy("resume_generation"),
+            EngineControlPolicy::RegisterAfter
+        );
         assert_eq!(
             engine_control_policy("resume_memory_occupation"),
             EngineControlPolicy::RegisterAfter

lib/bindings/python/rust/backend.rs

@@ -460,6 +460,9 @@ impl WorkerConfig {
                 structural_tag_schema: st_schema,
                 runtime: runtime.map(|r| r.inner).unwrap_or_default(),
                 route_to_encoder,
+                // Python vLLM owns and serves its existing `.rl` endpoint.
+                // The shared Rust endpoint is opt-in for Rust sidecars only.
+                enable_rl: false,
                 media_decoder: media_decoder.map(|decoder| decoder.inner),
                 media_fetcher: media_fetcher.map(|fetcher| fetcher.inner),
             },

lib/mocker/servers/vllm/Cargo.toml

@@ -9,7 +9,7 @@ authors.workspace = true
 license.workspace = true
 homepage.workspace = true
 repository.workspace = true
-description = "Mocker-backed implementation of vLLM's native Generate gRPC API"
+description = "Mocker-backed implementation of vLLM's native gRPC API"
 readme = "README.md"
 
 [[bin]]
@@ -28,6 +28,7 @@ futures = { workspace = true }
 prost-types = { workspace = true }
 tokio = { workspace = true }
 tonic = { workspace = true }
+tonic-health = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 uuid = { workspace = true }

lib/mocker/servers/vllm/README.md

@@ -5,14 +5,9 @@ SPDX-License-Identifier: Apache-2.0
 
 # Mocker-backed vLLM gRPC server
 
-`dynamo-vllm-mocker-server` implements vLLM's native `Generate` and
-`GenerateStream` RPCs on CPU, using the Dynamo Mocker scheduler for batching,
-KV-capacity, prefix-cache, and timing behavior. Its primary purpose is fast,
-repeatable testing of `dynamo-vllm-sidecar` without a model or GPU.
+`dynamo-vllm-mocker-server` implements vLLM's native Inference and Control services plus standard gRPC health on CPU. It uses the Dynamo Mocker scheduler for batching, KV capacity, prefix cache, and timing behavior.
 
-The mock server temporarily imports the generated types exposed by
-`dynamo-vllm-sidecar`, whose proto is vendored unchanged from vLLM v0.25.1.
-Both consumers will move to vLLM's upstream package once it is published.
+The mock server imports the generated types exposed by `dynamo-vllm-sidecar`. The proto files are vendored unchanged from vLLM.
 
 ## Aggregated serving
 
@@ -29,8 +24,7 @@ Point the existing Dynamo sidecar at it:
 
 ```bash
 cargo run -p dynamo-vllm-sidecar --bin dynamo-vllm-sidecar -- \
-  --vllm-endpoint 127.0.0.1:50051 \
-  --model-path mocker-model
+  --vllm-endpoint 127.0.0.1:50051
 ```
 
 `--extra-engine-args` accepts inline JSON or a JSON file path. The values use
@@ -62,14 +56,16 @@ Then start one sidecar for each endpoint:
 
 ```bash
 cargo run -p dynamo-vllm-sidecar --bin dynamo-vllm-sidecar -- \
-  --vllm-endpoint 127.0.0.1:50051 --model-path mocker-model \
+  --vllm-endpoint 127.0.0.1:50051 \
   --disaggregation-mode prefill
 
 cargo run -p dynamo-vllm-sidecar --bin dynamo-vllm-sidecar -- \
-  --vllm-endpoint 127.0.0.1:50052 --model-path mocker-model \
+  --vllm-endpoint 127.0.0.1:50052 \
   --disaggregation-mode decode
 ```
 
+The sidecar discovers model identity through Control. Keep `--disaggregation-mode` for prefill and decode because the current discovery API does not report engine role.
+
 The prefill endpoint returns an opaque vLLM-shaped `kv_transfer_params`
 payload, and the decode endpoint validates that the sidecar forwarded it
 verbatim — including a non-rendezvous sentinel field, so a dropped opaque field