diff --git a/.github/workflows/copyright-check.ps1 b/.github/workflows/copyright-check.ps1
index 6827a8f21e85..eeb0d84692f7 100644
--- a/.github/workflows/copyright-check.ps1
+++ b/.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 "', '")']."
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 0e91f761e025..95503eac15ba 100644
--- a/.pre-commit-config.yaml
+++ b/.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
diff --git a/Cargo.lock b/Cargo.lock
index 71015670b83f..7b7fdf0d2aec 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3084,6 +3084,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tonic 0.13.1",
+ "tonic-health",
  "tracing",
  "tracing-subscriber",
  "uuid",
@@ -3108,6 +3109,7 @@ dependencies = [
  "tokio-util",
  "tonic 0.13.1",
  "tonic-build 0.13.1",
+ "tonic-health",
  "tracing",
 ]
 
diff --git a/components/src/dynamo/common/backend/engine.py b/components/src/dynamo/common/backend/engine.py
index 4963637c03ec..f972b385d28e 100644
--- a/components/src/dynamo/common/backend/engine.py
+++ b/components/src/dynamo/common/backend/engine.py
@@ -6,7 +6,7 @@
 import os
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Callable, Optional, TypedDict
 
 from typing_extensions import Required
@@ -137,9 +137,9 @@ class LlmRegistration:
 class EngineConfig:
     """Registration metadata returned by an engine's :meth:`start`.
 
-    The neutral fields (``model``, ``served_model_name``, ``runtime_data``)
-    apply to every modality; token-pipeline metadata lives in the optional
-    :attr:`llm` sub-record, which raw media engines leave ``None``.
+    The neutral fields (``model``, ``served_model_name``, ``model_aliases``,
+    ``runtime_data``) apply to every modality; token-pipeline metadata lives in
+    the optional :attr:`llm` sub-record, which raw media engines leave ``None``.
     """
 
     model: str
@@ -148,6 +148,8 @@ class EngineConfig:
     # Token-pipeline registration metadata (KV cache, DP, bootstrap).
     # ``Some`` for LLMEngines; ``None`` for RawEngines.
     llm: Optional[LlmRegistration] = None
+    # Kept after existing fields to preserve positional-constructor compatibility.
+    model_aliases: list[str] = field(default_factory=list)
 
 
 class BaseEngine(ABC):
diff --git a/components/src/dynamo/common/backend/tests/test_backend_bindings.py b/components/src/dynamo/common/backend/tests/test_backend_bindings.py
index c6403b75e6ef..93f40f3882fd 100644
--- a/components/src/dynamo/common/backend/tests/test_backend_bindings.py
+++ b/components/src/dynamo/common/backend/tests/test_backend_bindings.py
@@ -67,6 +67,7 @@ def test_engine_config_required_model_only():
     cfg = backend.EngineConfig(model="m1")
     assert cfg.model == "m1"
     assert cfg.served_model_name is None
+    assert cfg.model_aliases == []
     assert cfg.llm is None
 
 
@@ -74,6 +75,7 @@ def test_engine_config_full_kwargs_round_trip_through_getters():
     cfg = backend.EngineConfig(
         model="m2",
         served_model_name="m2-serving",
+        model_aliases=["m2-alias"],
         runtime_data={"sglang_worker_group_id": "group-a"},
         llm=backend.LlmRegistration(
             context_length=2048,
@@ -85,6 +87,7 @@ def test_engine_config_full_kwargs_round_trip_through_getters():
     )
     assert cfg.model == "m2"
     assert cfg.served_model_name == "m2-serving"
+    assert cfg.model_aliases == ["m2-alias"]
     assert cfg.runtime_data == {"sglang_worker_group_id": "group-a"}
     llm = cfg.llm
     assert llm.context_length == 2048
diff --git a/lib/backend-common/examples/mocker/src/engine.rs b/lib/backend-common/examples/mocker/src/engine.rs
index 6c9795678780..9fa42059f497 100644
--- a/lib/backend-common/examples/mocker/src/engine.rs
+++ b/lib/backend-common/examples/mocker/src/engine.rs
@@ -353,6 +353,7 @@ impl LLMEngine for MockerBackend {
         Ok(EngineConfig {
             model: self.model_name.clone(),
             served_model_name: Some(self.model_name.clone()),
+            model_aliases: Vec::new(),
             runtime_data: Default::default(),
             llm: Some(LlmRegistration {
                 context_length: Some(self.context_length),
diff --git a/lib/backend-common/src/engine.rs b/lib/backend-common/src/engine.rs
index c7368738db29..e1046b4deb88 100644
--- a/lib/backend-common/src/engine.rs
+++ b/lib/backend-common/src/engine.rs
@@ -145,15 +145,17 @@ pub struct LlmRegistration {
 ///
 /// `Worker` consumes this to build a `ModelDeploymentCard` and register the
 /// model with discovery. The neutral fields (`model`, `served_model_name`,
-/// `runtime_data`) apply to every modality; the token-pipeline metadata lives
-/// in the optional [`llm`](Self::llm) sub-record, which raw media engines
-/// leave `None`.
+/// `model_aliases`, `runtime_data`) apply to every modality; the token-pipeline
+/// metadata lives in the optional [`llm`](Self::llm) sub-record, which raw
+/// media engines leave `None`.
 #[derive(Clone, Debug, Default)]
 pub struct EngineConfig {
     /// Canonical model identifier (e.g. HF repo name).
     pub model: String,
     /// Public-facing model name advertised to clients. Defaults to `model`.
     pub served_model_name: Option<String>,
+    /// Additional public-facing model names accepted by the engine.
+    pub model_aliases: Vec<String>,
     /// Engine-specific metadata copied into `ModelRuntimeConfig.runtime_data`.
     pub runtime_data: HashMap<String, serde_json::Value>,
     /// Token-pipeline registration metadata (KV cache, DP, bootstrap).
diff --git a/lib/backend-common/src/worker.rs b/lib/backend-common/src/worker.rs
index 82403f7e219f..1aa1fa6d82b3 100644
--- a/lib/backend-common/src/worker.rs
+++ b/lib/backend-common/src/worker.rs
@@ -1683,6 +1683,7 @@ async fn build_local_model(
     let mut builder = LocalModelBuilder::default();
     builder
         .model_name(served_name)
+        .model_aliases(engine_config.model_aliases.clone())
         .kv_cache_block_size(llm.kv_cache_block_size)
         .custom_template_path(config.custom_jinja_template.clone())
         .media_decoder(config.media_decoder.clone())
@@ -2030,6 +2031,7 @@ mod tests {
         };
         let engine_config = EngineConfig {
             model: "media-config-test".to_string(),
+            model_aliases: vec!["media-alias".to_string()],
             ..EngineConfig::default()
         };
 
@@ -2039,6 +2041,7 @@ mod tests {
 
         assert!(local_model.card().media_decoder.is_some());
         assert!(local_model.card().media_fetcher.is_some());
+        assert_eq!(local_model.card().aliases, ["media-alias"]);
     }
 
     #[test]
diff --git a/lib/bindings/python/rust/backend.rs b/lib/bindings/python/rust/backend.rs
index d691238d9db5..803318c3008a 100644
--- a/lib/bindings/python/rust/backend.rs
+++ b/lib/bindings/python/rust/backend.rs
@@ -134,7 +134,7 @@ impl From<DisaggregationMode> for RsDisaggregationMode {
 // EngineConfig — mirror of `dynamo_backend_common::EngineConfig`.
 //
 // Engines may return either this pyclass or any object with the canonical
-// attributes `model` / `served_model_name` / `runtime_data` / `llm`; the
+// attributes `model` / `served_model_name` / `model_aliases` / `runtime_data` / `llm`; the
 // bridge's `start()` extraction accepts both. Note `llm` is a nested record
 // (LlmRegistration), NOT flat fields — an object exposing flat `context_length`
 // etc. (the pre-split shape) registers with `llm=None`, i.e. no KV/DP/bootstrap
@@ -236,12 +236,13 @@ pub struct EngineConfig {
 #[pymethods]
 impl EngineConfig {
     #[new]
-    #[pyo3(signature = (model, served_model_name = None, runtime_data = None, llm = None))]
+    #[pyo3(signature = (model, served_model_name = None, runtime_data = None, llm = None, model_aliases = None))]
     fn new(
         model: String,
         served_model_name: Option<String>,
         runtime_data: Option<&Bound<'_, PyDict>>,
         llm: Option<LlmRegistration>,
+        model_aliases: Option<Vec<String>>,
     ) -> PyResult<Self> {
         let runtime_data = runtime_data
             .map(|dict| depythonize::<HashMap<String, serde_json::Value>>(dict))
@@ -253,6 +254,7 @@ impl EngineConfig {
             inner: RsEngineConfig {
                 model,
                 served_model_name,
+                model_aliases: model_aliases.unwrap_or_default(),
                 runtime_data,
                 llm: llm.map(|l| l.inner),
             },
@@ -268,6 +270,10 @@ impl EngineConfig {
         self.inner.served_model_name.as_deref()
     }
     #[getter]
+    fn model_aliases(&self) -> &[String] {
+        &self.inner.model_aliases
+    }
+    #[getter]
     fn llm(&self) -> Option<LlmRegistration> {
         self.inner
             .llm
@@ -862,6 +868,7 @@ impl PyEngineCore {
             Ok(RsEngineConfig {
                 model: bound.getattr("model")?.extract()?,
                 served_model_name: opt_attr::<String>(bound, "served_model_name")?,
+                model_aliases: opt_attr::<Vec<String>>(bound, "model_aliases")?.unwrap_or_default(),
                 runtime_data: match bound.getattr("runtime_data") {
                     Ok(value) if !value.is_none() => depythonize(&value).map_err(to_pyerr)?,
                     Ok(_) => HashMap::new(),
diff --git a/lib/bindings/python/src/dynamo/_core.pyi b/lib/bindings/python/src/dynamo/_core.pyi
index d7d034d68f5f..aecaa27bf58b 100644
--- a/lib/bindings/python/src/dynamo/_core.pyi
+++ b/lib/bindings/python/src/dynamo/_core.pyi
@@ -3324,12 +3324,15 @@ class backend:
             served_model_name: Optional[str] = None,
             runtime_data: Optional[Dict[str, Any]] = None,
             llm: Optional["backend.LlmRegistration"] = None,
+            model_aliases: Optional[List[str]] = None,
         ) -> None: ...
         @property
         def model(self) -> str: ...
         @property
         def served_model_name(self) -> Optional[str]: ...
         @property
+        def model_aliases(self) -> List[str]: ...
+        @property
         def runtime_data(self) -> Dict[str, Any]: ...
         @property
         def llm(self) -> Optional["backend.LlmRegistration"]: ...
diff --git a/lib/mocker/servers/vllm/Cargo.toml b/lib/mocker/servers/vllm/Cargo.toml
index 95275eea4cab..17bcd78bf655 100644
--- a/lib/mocker/servers/vllm/Cargo.toml
+++ b/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 }
diff --git a/lib/mocker/servers/vllm/README.md b/lib/mocker/servers/vllm/README.md
index e9d85ab5e42c..e2363b227421 100644
--- a/lib/mocker/servers/vllm/README.md
+++ b/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
diff --git a/lib/mocker/servers/vllm/src/main.rs b/lib/mocker/servers/vllm/src/main.rs
index 4f4e279101a9..2e512c120d40 100644
--- a/lib/mocker/servers/vllm/src/main.rs
+++ b/lib/mocker/servers/vllm/src/main.rs
@@ -7,20 +7,20 @@ use anyhow::Context;
 use clap::Parser;
 use dynamo_mocker::common::protocols::MockEngineArgs;
 use dynamo_vllm_mocker::{MockerServerConfig, ServerMode, VllmMockerService};
-use dynamo_vllm_sidecar::proto::generate_server::GenerateServer;
+use dynamo_vllm_sidecar::proto::control_server::ControlServer;
+use dynamo_vllm_sidecar::proto::inference_server::InferenceServer;
 
 #[derive(Parser, Debug)]
 #[command(
     name = "dynamo-vllm-mocker-server",
-    about = "Run a CPU-only, Mocker-backed implementation of vLLM's native Generate gRPC API"
+    about = "Run a CPU-only, Mocker-backed implementation of vLLM's native gRPC API"
 )]
 struct Args {
     /// Address on which to expose the vLLM-compatible gRPC service.
     #[arg(long, default_value = "127.0.0.1:50051")]
     listen: SocketAddr,
 
-    /// Model name accepted in Generate requests. The empty model used by the
-    /// Dynamo vLLM sidecar is always accepted.
+    /// Model name exposed by the mock server.
     #[arg(long, default_value = "mocker-model")]
     model: String,
 
@@ -81,8 +81,17 @@ async fn main() -> anyhow::Result<()> {
         mode = %service.config().mode,
         "starting Mocker-backed vLLM gRPC server"
     );
+    let (health, health_service) = tonic_health::server::health_reporter();
+    health
+        .set_serving::<ControlServer<VllmMockerService>>()
+        .await;
+    health
+        .set_serving::<InferenceServer<VllmMockerService>>()
+        .await;
     tonic::transport::Server::builder()
-        .add_service(GenerateServer::new(service))
+        .add_service(InferenceServer::new(service.clone()))
+        .add_service(ControlServer::new(service))
+        .add_service(health_service)
         .serve_with_shutdown(args.listen, async {
             let _ = tokio::signal::ctrl_c().await;
         })
diff --git a/lib/mocker/servers/vllm/src/server.rs b/lib/mocker/servers/vllm/src/server.rs
index 687d7b11a98f..c41f8bd87a01 100644
--- a/lib/mocker/servers/vllm/src/server.rs
+++ b/lib/mocker/servers/vllm/src/server.rs
@@ -7,7 +7,7 @@ use std::sync::Arc;
 
 use clap::ValueEnum;
 use dynamo_mocker::common::protocols::{EngineType, MockEngineArgs, OutputSignal, WorkerType};
-use dynamo_mocker::live::{LiveEngine, LiveRequest};
+use dynamo_mocker::live::{LiveEngine, LiveRequest, stable_request_uuid};
 use dynamo_mocker::scheduler::MockerMetrics;
 use dynamo_vllm_sidecar::proto as pb;
 use futures::Stream;
@@ -60,10 +60,12 @@ impl Default for MockerServerConfig {
     }
 }
 
-/// vLLM-compatible Generate service driven by one shared Mocker scheduler.
+/// Mocker-backed vLLM services.
 #[derive(Clone)]
 pub struct VllmMockerService {
     config: Arc<MockerServerConfig>,
+    model_info: Arc<pb::ModelInfo>,
+    server_info: Arc<pb::ServerInfo>,
     engine: LiveEngine,
     request_permits: Arc<Semaphore>,
 }
@@ -84,8 +86,57 @@ impl VllmMockerService {
             "Mocker worker_type must be aggregated; use the server mode for the emulated wire role"
         );
         let max_concurrent_requests = config.max_concurrent_requests;
+        let model_info = pb::ModelInfo {
+            model_id: config.model.clone(),
+            served_model_name: config.model.clone(),
+            served_model_aliases: Vec::new(),
+            supports_text_input: false,
+            supports_token_ids_input: true,
+            supports_multimodal: false,
+            reasoning_parser: String::new(),
+            tool_call_parser: String::new(),
+        };
+        let server_info = pb::ServerInfo {
+            engine_version: env!("CARGO_PKG_VERSION").to_string(),
+            api_version: "vllm".to_string(),
+            instance_id: format!("dynamo-vllm-mocker-{}", config.mode),
+            parallelism: Some(pb::ParallelismInfo {
+                tensor_parallel_size: 1,
+                pipeline_parallel_size: 1,
+                data_parallel_size: engine_args.dp_size,
+                data_parallel_rank: DP_RANK,
+                decode_context_parallel_size: 1,
+            }),
+            max_model_len: engine_args
+                .max_model_len
+                .map(u32::try_from)
+                .transpose()
+                .map_err(|_| anyhow::anyhow!("max_model_len exceeds the Control API range"))?
+                .unwrap_or_default(),
+            kv_block_size: u32::try_from(engine_args.block_size)
+                .map_err(|_| anyhow::anyhow!("block_size exceeds the Control API range"))?,
+            total_kv_blocks: u64::try_from(engine_args.num_gpu_blocks)
+                .map_err(|_| anyhow::anyhow!("num_gpu_blocks exceeds the Control API range"))?,
+            max_running_requests: engine_args
+                .max_num_seqs
+                .map(u64::try_from)
+                .transpose()
+                .map_err(|_| anyhow::anyhow!("max_num_seqs exceeds the Control API range"))?
+                .unwrap_or_default(),
+            max_batched_tokens: engine_args
+                .max_num_batched_tokens
+                .map(u64::try_from)
+                .transpose()
+                .map_err(|_| {
+                    anyhow::anyhow!("max_num_batched_tokens exceeds the Control API range")
+                })?
+                .unwrap_or_default(),
+            supports_explicit_data_parallel_rank: true,
+        };
         Ok(Self {
             config: Arc::new(config),
+            model_info: Arc::new(model_info),
+            server_info: Arc::new(server_info),
             engine: LiveEngine::start(engine_args, DP_RANK)?,
             request_permits: Arc::new(Semaphore::new(max_concurrent_requests)),
         })
@@ -125,7 +176,7 @@ impl VllmMockerService {
 }
 
 #[tonic::async_trait]
-impl pb::generate_server::Generate for VllmMockerService {
+impl pb::inference_server::Inference for VllmMockerService {
     type GenerateStreamStream =
         Pin<Box<dyn Stream<Item = Result<pb::GenerateResponse, Status>> + Send + 'static>>;
 
@@ -209,6 +260,45 @@ impl pb::generate_server::Generate for VllmMockerService {
     }
 }
 
+#[tonic::async_trait]
+impl pb::control_server::Control for VllmMockerService {
+    async fn get_server_info(
+        &self,
+        _request: Request<pb::GetServerInfoRequest>,
+    ) -> Result<Response<pb::ServerInfo>, Status> {
+        Ok(Response::new((*self.server_info).clone()))
+    }
+
+    async fn get_model_info(
+        &self,
+        _request: Request<pb::GetModelInfoRequest>,
+    ) -> Result<Response<pb::ModelInfo>, Status> {
+        Ok(Response::new((*self.model_info).clone()))
+    }
+
+    async fn abort(
+        &self,
+        request: Request<pb::AbortRequest>,
+    ) -> Result<Response<pb::AbortResponse>, Status> {
+        for request_id in request.into_inner().request_ids {
+            self.engine
+                .cancel(stable_request_uuid(self.config.seed, &request_id))
+                .await
+                .map_err(|error| Status::internal(format!("Mocker abort failed: {error}")))?;
+        }
+        Ok(Response::new(pb::AbortResponse {}))
+    }
+
+    async fn get_kv_event_sources(
+        &self,
+        _request: Request<pb::GetKvEventSourcesRequest>,
+    ) -> Result<Response<pb::GetKvEventSourcesResponse>, Status> {
+        Ok(Response::new(pb::GetKvEventSourcesResponse {
+            sources: Vec::new(),
+        }))
+    }
+}
+
 fn checked_token(signal: &OutputSignal) -> BoxedStatusResult<u32> {
     if signal.rejected {
         return Err(
diff --git a/lib/mocker/servers/vllm/src/server_request.rs b/lib/mocker/servers/vllm/src/server_request.rs
index 42021fd7c52e..61e5736b72a9 100644
--- a/lib/mocker/servers/vllm/src/server_request.rs
+++ b/lib/mocker/servers/vllm/src/server_request.rs
@@ -75,6 +75,14 @@ impl PreparedRequest {
             ))
             .into());
         }
+        if let Some(rank) = request.data_parallel_rank
+            && rank != DP_RANK
+        {
+            return Err(Status::invalid_argument(format!(
+                "data_parallel_rank {rank} is not served; expected {DP_RANK}"
+            ))
+            .into());
+        }
         let mut prompt_tokens = match request.prompt.take() {
             Some(pb::generate_request::Prompt::TokenIds(tokens)) => tokens.ids,
             Some(pb::generate_request::Prompt::Text(_)) => {
@@ -291,6 +299,7 @@ impl PreparedRequest {
                 finish_reason: pb::finish_info::FinishReason::Length as i32,
                 stop_reason: None,
                 kv_transfer_params: (self.mode == ServerMode::Prefill).then(|| self.handoff()),
+                ec_transfer_params: None,
             }),
         }
     }
diff --git a/lib/mocker/servers/vllm/src/server_tests.rs b/lib/mocker/servers/vllm/src/server_tests.rs
index 97cadccfd183..0260360e3917 100644
--- a/lib/mocker/servers/vllm/src/server_tests.rs
+++ b/lib/mocker/servers/vllm/src/server_tests.rs
@@ -219,7 +219,7 @@ async fn unary_generate_maps_capacity_rejection_to_resource_exhausted() {
         ids: vec![1, 2, 3, 4, 5],
     }));
 
-    let error = pb::generate_server::Generate::generate(&service, Request::new(oversized))
+    let error = pb::inference_server::Inference::generate(&service, Request::new(oversized))
         .await
         .unwrap_err();
     assert_eq!(error.code(), tonic::Code::ResourceExhausted);
@@ -245,18 +245,18 @@ async fn concurrent_request_limit_rejects_a_stalled_stream() {
     let mut first_request = request("stalled");
     first_request.stopping.as_mut().unwrap().max_new_tokens = 100;
     let first =
-        pb::generate_server::Generate::generate_stream(&service, Request::new(first_request))
+        pb::inference_server::Inference::generate_stream(&service, Request::new(first_request))
             .await
             .unwrap();
 
     let mut queued_request = request("queued");
     queued_request.stopping.as_mut().unwrap().max_new_tokens = 100;
     let queued =
-        pb::generate_server::Generate::generate_stream(&service, Request::new(queued_request))
+        pb::inference_server::Inference::generate_stream(&service, Request::new(queued_request))
             .await
             .unwrap();
 
-    let error = match pb::generate_server::Generate::generate_stream(
+    let error = match pb::inference_server::Inference::generate_stream(
         &service,
         Request::new(request("rejected")),
     )
@@ -312,7 +312,7 @@ async fn unary_generate_accumulates_output_and_terminal_metadata() {
     let service = VllmMockerService::new(MockerServerConfig::default(), admitting_args()).unwrap();
 
     let response =
-        pb::generate_server::Generate::generate(&service, Request::new(request("unary")))
+        pb::inference_server::Inference::generate(&service, Request::new(request("unary")))
             .await
             .unwrap()
             .into_inner();
@@ -355,7 +355,7 @@ async fn streaming_generate_maps_capacity_rejection_to_resource_exhausted() {
     }));
 
     let mut stream =
-        pb::generate_server::Generate::generate_stream(&service, Request::new(oversized))
+        pb::inference_server::Inference::generate_stream(&service, Request::new(oversized))
             .await
             .expect("streaming RPC opens before the scheduler rejects")
             .into_inner();
@@ -384,10 +384,11 @@ async fn streaming_survives_a_producer_that_outruns_a_stalled_consumer() {
     let mut bursty = request("bursty");
     bursty.stopping.as_mut().unwrap().max_new_tokens = 50;
 
-    let mut stream = pb::generate_server::Generate::generate_stream(&service, Request::new(bursty))
-        .await
-        .unwrap()
-        .into_inner();
+    let mut stream =
+        pb::inference_server::Inference::generate_stream(&service, Request::new(bursty))
+            .await
+            .unwrap()
+            .into_inner();
 
     // Stall the consumer so the instant producer fills and overflows the fixed
     // per-request buffer before we read anything.
diff --git a/lib/mocker/servers/vllm/tests/sidecar.rs b/lib/mocker/servers/vllm/tests/sidecar.rs
index ee60ea908858..d65a3214e8a3 100644
--- a/lib/mocker/servers/vllm/tests/sidecar.rs
+++ b/lib/mocker/servers/vllm/tests/sidecar.rs
@@ -10,7 +10,8 @@ use dynamo_backend_common::{
 use dynamo_mocker::common::protocols::MockEngineArgs;
 use dynamo_vllm_mocker::{MockerServerConfig, ServerMode, VllmMockerService};
 use dynamo_vllm_sidecar::VllmSidecarEngine;
-use dynamo_vllm_sidecar::proto::generate_server::GenerateServer;
+use dynamo_vllm_sidecar::proto::control_server::ControlServer;
+use dynamo_vllm_sidecar::proto::inference_server::InferenceServer;
 use futures::StreamExt;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
@@ -35,10 +36,20 @@ impl RunningServer {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let address = listener.local_addr().unwrap();
         let (shutdown, shutdown_rx) = oneshot::channel();
-        let server_service = service.clone();
+        let inference_service = service.clone();
+        let control_service = service.clone();
+        let (health, health_service) = tonic_health::server::health_reporter();
+        health
+            .set_serving::<ControlServer<VllmMockerService>>()
+            .await;
+        health
+            .set_serving::<InferenceServer<VllmMockerService>>()
+            .await;
         tokio::spawn(async move {
             tonic::transport::Server::builder()
-                .add_service(GenerateServer::new(server_service))
+                .add_service(InferenceServer::new(inference_service))
+                .add_service(ControlServer::new(control_service))
+                .add_service(health_service)
                 .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async {
                     let _ = shutdown_rx.await;
                 })
@@ -73,13 +84,11 @@ fn fast_engine_args() -> MockEngineArgs {
         .unwrap()
 }
 
-fn sidecar(endpoint: &str, mode: DisaggregationMode) -> VllmSidecarEngine {
+async fn sidecar(endpoint: &str, mode: DisaggregationMode) -> VllmSidecarEngine {
     let mut argv = vec![
         "dynamo-vllm-sidecar".to_string(),
         "--vllm-endpoint".to_string(),
         endpoint.to_string(),
-        "--model-path".to_string(),
-        "mocker-model".to_string(),
         "--grpc-connections".to_string(),
         "1".to_string(),
         "--grpc-startup-deadline-secs".to_string(),
@@ -90,7 +99,11 @@ fn sidecar(endpoint: &str, mode: DisaggregationMode) -> VllmSidecarEngine {
     if mode != DisaggregationMode::Aggregated {
         argv.extend(["--disaggregation-mode".to_string(), mode.to_string()]);
     }
-    VllmSidecarEngine::from_args(Some(argv)).unwrap().0
+    tokio::task::spawn_blocking(move || VllmSidecarEngine::from_args(Some(argv)))
+        .await
+        .unwrap()
+        .unwrap()
+        .0
 }
 
 fn request(max_tokens: u32) -> PreprocessedRequest {
@@ -132,7 +145,7 @@ async fn collect(
 #[tokio::test]
 async fn sidecar_streams_mocker_tokens_logprobs_and_usage() {
     let server = RunningServer::start(ServerMode::Aggregated, fast_engine_args()).await;
-    let engine = sidecar(&server.endpoint, DisaggregationMode::Aggregated);
+    let engine = sidecar(&server.endpoint, DisaggregationMode::Aggregated).await;
     engine.start(0).await.unwrap();
 
     let outputs = collect(&engine, request(3)).await;
@@ -160,8 +173,8 @@ async fn sidecar_streams_mocker_tokens_logprobs_and_usage() {
 async fn prefill_handoff_round_trips_through_a_decode_server() {
     let prefill_server = RunningServer::start(ServerMode::Prefill, fast_engine_args()).await;
     let decode_server = RunningServer::start(ServerMode::Decode, fast_engine_args()).await;
-    let prefill = sidecar(&prefill_server.endpoint, DisaggregationMode::Prefill);
-    let decode = sidecar(&decode_server.endpoint, DisaggregationMode::Decode);
+    let prefill = sidecar(&prefill_server.endpoint, DisaggregationMode::Prefill).await;
+    let decode = sidecar(&decode_server.endpoint, DisaggregationMode::Decode).await;
     prefill.start(0).await.unwrap();
     decode.start(1).await.unwrap();
 
@@ -199,7 +212,7 @@ async fn dropping_sidecar_stream_cancels_mocker_work() {
     let mut args = fast_engine_args();
     args.speedup_ratio = 0.1;
     let server = RunningServer::start(ServerMode::Aggregated, args).await;
-    let engine = sidecar(&server.endpoint, DisaggregationMode::Aggregated);
+    let engine = sidecar(&server.endpoint, DisaggregationMode::Aggregated).await;
     engine.start(0).await.unwrap();
 
     let context = dynamo_backend_common::testing::mock_context();
diff --git a/lib/sidecar/common/src/args.rs b/lib/sidecar/common/src/args.rs
index 94a7a12a6395..7f591196bff6 100644
--- a/lib/sidecar/common/src/args.rs
+++ b/lib/sidecar/common/src/args.rs
@@ -43,7 +43,7 @@ pub struct GrpcTransportArgs {
     )]
     pub grpc_retry_interval_secs: u64,
 
-    /// Maximum total duration for sidecar gRPC startup.
+    /// Maximum duration for each sidecar gRPC startup phase.
     #[arg(
         long = "grpc-startup-deadline-secs",
         env = "DYN_SIDECAR_GRPC_STARTUP_DEADLINE_SECS",
diff --git a/lib/sidecar/sglang/src/engine.rs b/lib/sidecar/sglang/src/engine.rs
index 8b1bccc6df0a..5f1d408ac976 100644
--- a/lib/sidecar/sglang/src/engine.rs
+++ b/lib/sidecar/sglang/src/engine.rs
@@ -590,6 +590,7 @@ fn build_engine_config(
     Ok(EngineConfig {
         model: discovery.model_path.clone(),
         served_model_name: discovery.served_model_name.clone(),
+        model_aliases: Vec::new(),
         runtime_data,
         llm: Some(LlmRegistration {
             context_length: discovery.max_model_len,
diff --git a/lib/sidecar/trtllm/src/model.rs b/lib/sidecar/trtllm/src/model.rs
index 6d3567b4978f..b2b3795b62a5 100644
--- a/lib/sidecar/trtllm/src/model.rs
+++ b/lib/sidecar/trtllm/src/model.rs
@@ -26,6 +26,7 @@ impl ConfiguredModel {
         EngineConfig {
             model: self.source.clone(),
             served_model_name: None,
+            model_aliases: Vec::new(),
             runtime_data,
             llm: Some(LlmRegistration {
                 context_length: self.context_length,
diff --git a/lib/sidecar/vllm/Cargo.toml b/lib/sidecar/vllm/Cargo.toml
index 884f73eb8770..ea6603f2714f 100644
--- a/lib/sidecar/vllm/Cargo.toml
+++ b/lib/sidecar/vllm/Cargo.toml
@@ -36,6 +36,7 @@ tracing = { workspace = true }
 prost = { workspace = true }
 prost-types = { workspace = true }
 tonic = { workspace = true }
+tonic-health = { workspace = true }
 
 [build-dependencies]
 tonic-build = { workspace = true }
diff --git a/lib/sidecar/vllm/README.md b/lib/sidecar/vllm/README.md
index 987ea020bbd6..b25604b15c02 100644
--- a/lib/sidecar/vllm/README.md
+++ b/lib/sidecar/vllm/README.md
@@ -10,8 +10,13 @@ SPDX-License-Identifier: Apache-2.0
 > and not yet packaged for distribution. The manifests, flags, and behavior may
 > change without notice.
 
-`dynamo-vllm-sidecar` connects a Dynamo worker to vLLM's native gRPC
-`Generate` service. It is a standalone Rust executable.
+`dynamo-vllm-sidecar` connects a Dynamo worker to vLLM's native gRPC services:
+
+- `vllm.Inference` for generation
+- `vllm.Control` for model and server discovery
+- Standard gRPC health for startup readiness
+
+It is a standalone Rust executable.
 
 ## Supported
 
@@ -40,13 +45,18 @@ Start the Dynamo worker explicitly:
 
 ```bash
 dynamo-vllm-sidecar \
-  --vllm-endpoint 127.0.0.1:50051 \
-  --model-path Qwen/Qwen3-0.6B
+  --vllm-endpoint 127.0.0.1:50051
 ```
 
 Use `VLLM_GRPC_ENDPOINT` instead of `--vllm-endpoint` when the endpoint is
 provided through the environment.
 
+The sidecar discovers `model_id`, the served name, context length, KV capacity, and scheduler limits through `vllm.Control`. `model_id` must be readable locally or fetchable by Dynamo for tokenization and chat templates. Parser defaults are not advertised because the current inference protocol cannot preserve all parser-related request semantics.
+
+Data-parallel registration is omitted because Control reports global topology, not the rank range hosted by the connected frontend.
+
+Aggregated serving is the default. Set the existing `--disaggregation-mode` to `prefill` or `decode` only for non-aggregated deployments; the current Control API does not report engine role.
+
 The sidecar opens eight gRPC connections by default. This avoided
 connection-level throttling in high-concurrency sidecar tests. Override the
 pool size with `--grpc-connections` or `DYN_SIDECAR_GRPC_CONNECTIONS`.
@@ -59,8 +69,7 @@ corresponding `DYN_SIDECAR_GRPC_*` environment variables.
 
 ## Test without vLLM or a GPU
 
-Use the CPU-only `dynamo-vllm-mocker-server` to exercise this sidecar against
-the same `Generate` gRPC contract:
+Use the CPU-only `dynamo-vllm-mocker-server` to exercise the same Inference, Control, and health contracts:
 
 ```bash
 cargo run -p dynamo-vllm-mocker --bin dynamo-vllm-mocker-server -- \
@@ -69,8 +78,7 @@ cargo run -p dynamo-vllm-mocker --bin dynamo-vllm-mocker-server -- \
   --extra-engine-args '{"speedup_ratio":1000}'
 
 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
 ```
 
 See [`../../mocker/servers/vllm/README.md`](../../mocker/servers/vllm/README.md)
@@ -86,11 +94,7 @@ disaggregated prefill/decode with NIXL KV transfer.
 There is no published vLLM sidecar image yet, so you build and push your own from
 `Dockerfile` — the same pattern as the TensorRT-LLM and SGLang sidecars.
 
-> [!NOTE]
-> vLLM's `vllm-rs` exposes no gRPC health method, so the engine's probes can only
-> confirm the gRPC port accepts connections (not that the model is loaded) —
-> weaker than the TensorRT-LLM and SGLang sidecars, which make a real HealthCheck
-> RPC. The engine image must be a stock vLLM build that ships `vllm-rs` (v0.26.0+).
+The sidecar waits for both the Control and Inference services through the standard gRPC health API before registering the worker. The deployment manifests retain lightweight socket probes for container lifecycle monitoring. The engine image must include a `vllm-rs` build compatible with the vendored protocol.
 
 ### Prerequisites
 
diff --git a/lib/sidecar/vllm/build.rs b/lib/sidecar/vllm/build.rs
index d4255c891ce6..7f01c4e896f2 100644
--- a/lib/sidecar/vllm/build.rs
+++ b/lib/sidecar/vllm/build.rs
@@ -4,7 +4,11 @@
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     tonic_build::configure()
         .protoc_arg("--experimental_allow_proto3_optional")
-        .compile_protos(&["proto/vllm_grpc.proto"], &["proto"])?;
-    println!("cargo:rerun-if-changed=proto/vllm_grpc.proto");
+        .compile_protos(
+            &["proto/inference.proto", "proto/control.proto"],
+            &["proto"],
+        )?;
+    println!("cargo:rerun-if-changed=proto/inference.proto");
+    println!("cargo:rerun-if-changed=proto/control.proto");
     Ok(())
 }
diff --git a/lib/sidecar/vllm/deploy/agg.yaml b/lib/sidecar/vllm/deploy/agg.yaml
index a259a09f5c70..dc585738f60f 100644
--- a/lib/sidecar/vllm/deploy/agg.yaml
+++ b/lib/sidecar/vllm/deploy/agg.yaml
@@ -44,13 +44,9 @@ spec:
         - name: vllm-engine
           image: vllm/vllm-openai:v0.26.0
           restartPolicy: Always
-          # vllm-rs exposes no gRPC health method (grpc.health.v1 and any Generate
-          # HealthCheck both return UNIMPLEMENTED), so unlike the TensorRT-LLM and
-          # SGLang sidecars these probes can only confirm the gRPC port accepts a
-          # connection, not that the model is loaded. exec (not tcpSocket) because
-          # the loopback-bound listener is unreachable from the pod IP. The
-          # readinessProbe still drops the worker if the engine dies at runtime
-          # (port closes -> pod NotReady -> frontend stops routing to it).
+          # These lightweight lifecycle probes confirm that the loopback listener
+          # accepts connections. The Dynamo sidecar independently waits for both
+          # services through the standard gRPC health API before registration.
           startupProbe:
             exec:
               command:
@@ -98,10 +94,6 @@ spec:
           args:
           - --vllm-endpoint
           - 127.0.0.1:50051
-          # vLLM's gRPC does not expose model metadata; the sidecar needs the
-          # model for tokenization, chat templates, and registration.
-          - --model-path
-          - Qwen/Qwen3-0.6B
           envFrom:
           - secretRef:
               name: hf-token-secret
diff --git a/lib/sidecar/vllm/deploy/disagg.yaml b/lib/sidecar/vllm/deploy/disagg.yaml
index ece70772420e..a029987b15b2 100644
--- a/lib/sidecar/vllm/deploy/disagg.yaml
+++ b/lib/sidecar/vllm/deploy/disagg.yaml
@@ -67,8 +67,8 @@ spec:
             runAsUser: 0
             capabilities:
               add: ["IPC_LOCK", "SYS_RESOURCE"]
-          # vllm-rs has no gRPC health method (see agg.yaml), so these probes only
-          # confirm the port accepts a connection. readiness drops a dead engine.
+          # The socket probes cover container lifecycle. The Dynamo sidecar also
+          # checks both services through the standard gRPC health API.
           startupProbe:
             exec:
               command:
@@ -137,8 +137,6 @@ spec:
           args:
           - --vllm-endpoint
           - 127.0.0.1:50051
-          - --model-path
-          - Qwen/Qwen3-0.6B
           - --disaggregation-mode
           - prefill
           envFrom:
@@ -163,8 +161,8 @@ spec:
             runAsUser: 0
             capabilities:
               add: ["IPC_LOCK", "SYS_RESOURCE"]
-          # vllm-rs has no gRPC health method (see agg.yaml), so these probes only
-          # confirm the port accepts a connection. readiness drops a dead engine.
+          # The socket probes cover container lifecycle. The Dynamo sidecar also
+          # checks both services through the standard gRPC health API.
           startupProbe:
             exec:
               command:
@@ -230,8 +228,6 @@ spec:
           args:
           - --vllm-endpoint
           - 127.0.0.1:50051
-          - --model-path
-          - Qwen/Qwen3-0.6B
           - --disaggregation-mode
           - decode
           envFrom:
diff --git a/lib/sidecar/vllm/launch/agg.sh b/lib/sidecar/vllm/launch/agg.sh
index b06fcc35d76d..f37b3e0929e5 100755
--- a/lib/sidecar/vllm/launch/agg.sh
+++ b/lib/sidecar/vllm/launch/agg.sh
@@ -91,7 +91,6 @@ vllm-rs serve "$MODEL" \
 
 DYN_SYSTEM_PORT=${DYN_SYSTEM_PORT:-8081} \
     dynamo-vllm-sidecar \
-    --vllm-endpoint "127.0.0.1:${VLLM_GRPC_PORT}" \
-    --model-path "$MODEL" &
+    --vllm-endpoint "127.0.0.1:${VLLM_GRPC_PORT}" &
 
 wait_any_exit
diff --git a/lib/sidecar/vllm/launch/disagg.sh b/lib/sidecar/vllm/launch/disagg.sh
index 3ea63ccf5cff..5f40990aa285 100755
--- a/lib/sidecar/vllm/launch/disagg.sh
+++ b/lib/sidecar/vllm/launch/disagg.sh
@@ -125,7 +125,6 @@ OTEL_SERVICE_NAME=dynamo-worker-decode \
 DYN_SYSTEM_PORT="${DYN_SYSTEM_PORT1:-8081}" \
     dynamo-vllm-sidecar \
     --vllm-endpoint "127.0.0.1:${VLLM_DECODE_GRPC_PORT}" \
-    --model-path "$MODEL" \
     --disaggregation-mode decode &
 
 # Register prefill separately so the frontend routes each disaggregated stage.
@@ -133,7 +132,6 @@ OTEL_SERVICE_NAME=dynamo-worker-prefill \
 DYN_SYSTEM_PORT="${DYN_SYSTEM_PORT2:-8082}" \
     dynamo-vllm-sidecar \
     --vllm-endpoint "127.0.0.1:${VLLM_PREFILL_GRPC_PORT}" \
-    --model-path "$MODEL" \
     --component prefill \
     --disaggregation-mode prefill &
 
diff --git a/lib/sidecar/vllm/proto/README.md b/lib/sidecar/vllm/proto/README.md
index 33b612c99e99..bab4a03cc097 100644
--- a/lib/sidecar/vllm/proto/README.md
+++ b/lib/sidecar/vllm/proto/README.md
@@ -5,12 +5,9 @@ SPDX-License-Identifier: Apache-2.0
 
 # Vendored vLLM protocol
 
-- Source: `rust/proto/vllm_grpc.proto`
-- Release: `v0.25.1`
-- Commit: `752a3a504485790a2e8491cacbb35c137339ad34`
-- SHA-256: `7cccd0e1b2e54f189550e1090cc80321fc2bbd188c98a4e22701b28fdeb177b6`
+- Source: [`rust/proto/inference.proto`](https://github.com/connorcarpenter15/vllm/blob/2d2c3af18c52e8e4efa4b0b4903843b15c0dba0e/rust/proto/inference.proto) and [`rust/proto/control.proto`](https://github.com/connorcarpenter15/vllm/blob/2d2c3af18c52e8e4efa4b0b4903843b15c0dba0e/rust/proto/control.proto)
+- Commit: `2d2c3af18c52e8e4efa4b0b4903843b15c0dba0e`
+- `inference.proto` SHA-256: `a0d196dc240683e1c09abb54f324d4428d0c122a6802b44916ad2d96b491b06c`
+- `control.proto` SHA-256: `cd4e7a8043f19d05929a2f59f5a5442894a037ef2d65832d3f7099992b1f1dbd`
 
-The file is copied without modification. Update the revision and checksum when
-updating the protocol. `dynamo-vllm-sidecar` generates and temporarily exports
-these types for `dynamo-vllm-mocker-server`; both consumers will move to the
-upstream package once vLLM publishes it.
+The files are copied without modification. Update the revision and checksums together. `dynamo-vllm-sidecar` generates and temporarily exports these types for `dynamo-vllm-mocker-server`.
diff --git a/lib/sidecar/vllm/proto/control.proto b/lib/sidecar/vllm/proto/control.proto
new file mode 100644
index 000000000000..d2ec9da4e7cc
--- /dev/null
+++ b/lib/sidecar/vllm/proto/control.proto
@@ -0,0 +1,78 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+syntax = "proto3";
+package vllm;
+
+service Control {
+  rpc GetServerInfo (GetServerInfoRequest) returns (ServerInfo) {}
+  rpc GetModelInfo (GetModelInfoRequest) returns (ModelInfo) {}
+  rpc Abort (AbortRequest) returns (AbortResponse) {}
+  rpc GetKvEventSources (GetKvEventSourcesRequest) returns (GetKvEventSourcesResponse) {}
+}
+
+message GetServerInfoRequest {}
+
+message ServerInfo {
+  string engine_version = 1;
+  string api_version = 2;
+  string instance_id = 3;
+  ParallelismInfo parallelism = 4;
+  uint32 max_model_len = 5;
+  uint32 kv_block_size = 6;
+  uint64 total_kv_blocks = 7;
+  uint64 max_running_requests = 8;
+  uint64 max_batched_tokens = 9;
+  // GenerateRequest.data_parallel_rank is honored by this server. Clients
+  // that require deterministic rank routing must fail closed when this is
+  // false, because older servers accept and silently discard the field.
+  bool supports_explicit_data_parallel_rank = 10;
+}
+
+message ParallelismInfo {
+  uint32 tensor_parallel_size = 1;
+  uint32 pipeline_parallel_size = 2;
+  uint32 data_parallel_size = 3;
+  uint32 data_parallel_rank = 4;
+  uint32 decode_context_parallel_size = 5;
+}
+
+message GetModelInfoRequest {}
+
+message ModelInfo {
+  string model_id = 1;
+  string served_model_name = 2;
+  repeated string served_model_aliases = 3;
+
+  bool supports_text_input = 20;
+  bool supports_token_ids_input = 21;
+  bool supports_multimodal = 23;
+  string reasoning_parser = 24;
+  string tool_call_parser = 25;
+}
+
+message AbortRequest {
+  repeated string request_ids = 1;
+}
+
+message AbortResponse {}
+
+// ======================================================================================
+// KV discovery
+// ======================================================================================
+
+message GetKvEventSourcesRequest {}
+message GetKvEventSourcesResponse { repeated KvEventSource sources = 1; }
+
+message KvEventSource {
+  string transport = 1;
+  string endpoint = 2;
+  string topic = 3;
+  string replay_endpoint = 4;
+  optional uint32 data_parallel_rank = 5;
+  string encoding = 6;
+  uint32 schema_version = 7;
+  uint32 buffer_steps = 8;
+  uint32 hwm = 9;
+  uint32 max_queue_size = 10;
+}
diff --git a/lib/sidecar/vllm/proto/vllm_grpc.proto b/lib/sidecar/vllm/proto/inference.proto
similarity index 83%
rename from lib/sidecar/vllm/proto/vllm_grpc.proto
rename to lib/sidecar/vllm/proto/inference.proto
index 56c5f36442db..4acb6826504a 100644
--- a/lib/sidecar/vllm/proto/vllm_grpc.proto
+++ b/lib/sidecar/vllm/proto/inference.proto
@@ -1,10 +1,13 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
 syntax = "proto3";
 package vllm;
 
 import "google/protobuf/struct.proto";
 
 
-service Generate {
+service Inference {
   // Generates text given a prompt
   rpc Generate (GenerateRequest) returns (GenerateResponse) {}
   // Generates text given a prompt, streaming the outputs
@@ -42,6 +45,14 @@ message GenerateRequest {
   uint32 truncate_prompt_tokens = 11;
 
   int32 priority = 12;
+
+  optional string session_id = 13;
+
+  // Multimodal inputs aligned with placeholder markers in token_ids.
+  repeated MediaItem media = 14;
+
+  // Global data-parallel rank advertised by the Control service.
+  optional uint32 data_parallel_rank = 15;
 }
 
 message RandomSampling {
@@ -107,6 +118,9 @@ message KVCacheParameters {
 
   // KV Connector transfer parameters
   google.protobuf.Struct kv_transfer_params = 3;
+
+  // Encoder cache connector transfer parameters
+  google.protobuf.Struct ec_transfer_params = 4;
 }
 
 // Controls which extra candidate tokens at each position should be returned
@@ -173,6 +187,7 @@ message FinishInfo {
 
   google.protobuf.Struct kv_transfer_params = 6;
   //uint64 seed = 7;
+  google.protobuf.Struct ec_transfer_params = 8;
 }
 
 // Info for candidate tokens other than the input/sampled
@@ -194,3 +209,24 @@ message TokenIds {
   repeated uint32 ids = 1;
 }
 
+// ======================================================================================
+// Media
+// ======================================================================================
+
+enum Modality {
+  MODALITY_UNSPECIFIED = 0;
+  MODALITY_IMAGE = 1;
+  MODALITY_VIDEO = 2;
+  MODALITY_AUDIO = 3;
+}
+
+message MediaItem {
+  Modality modality = 1;
+  oneof source {
+    string url = 2;       // http:// or https://
+    string data_uri = 3;  // data:
+    bytes raw_bytes = 4;
+  }
+  string mime_type = 5;
+  string uuid = 6;
+}
diff --git a/lib/sidecar/vllm/src/args.rs b/lib/sidecar/vllm/src/args.rs
index b589bab084bd..2dc8a67ee3bb 100644
--- a/lib/sidecar/vllm/src/args.rs
+++ b/lib/sidecar/vllm/src/args.rs
@@ -15,12 +15,4 @@ pub(crate) struct Args {
     /// vLLM gRPC endpoint as host:port or an http:// URL.
     #[arg(long, env = "VLLM_GRPC_ENDPOINT")]
     pub vllm_endpoint: String,
-
-    /// Hugging Face model ID or local path used by Dynamo for model-card
-    /// registration, tokenization, and chat templates. The released vLLM gRPC
-    /// API does not expose this metadata, so it cannot be inferred from the
-    /// endpoint. This flag is temporary and can be removed once vLLM exposes
-    /// model and tokenizer metadata over gRPC.
-    #[arg(long)]
-    pub model_path: String,
 }
diff --git a/lib/sidecar/vllm/src/client.rs b/lib/sidecar/vllm/src/client.rs
index 4ead90cac393..e3cf9c19e902 100644
--- a/lib/sidecar/vllm/src/client.rs
+++ b/lib/sidecar/vllm/src/client.rs
@@ -1,15 +1,23 @@
 // SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::time::Duration;
+
 use dynamo_backend_common::DynamoError;
 use dynamo_sidecar_common::{
     DEFAULT_MAX_GRPC_MESSAGE_SIZE, GrpcChannelPool, GrpcEndpoint, GrpcTransportConfig,
 };
+use tokio::time::{Instant, sleep_until, timeout_at};
+use tonic_health::pb::health_check_response::ServingStatus;
+use tonic_health::pb::{HealthCheckRequest, health_client::HealthClient};
 
 pub(crate) use dynamo_sidecar_common::{engine_shutdown, invalid_argument, status_to_dynamo};
 
 use crate::proto as pb;
 
+pub(crate) const CONTROL_SERVICE: &str = "vllm.Control";
+pub(crate) const INFERENCE_SERVICE: &str = "vllm.Inference";
+
 pub(crate) struct VllmClient {
     pool: GrpcChannelPool,
 }
@@ -18,8 +26,18 @@ impl VllmClient {
     pub(crate) async fn connect(
         endpoint: &GrpcEndpoint,
         transport: GrpcTransportConfig,
+        startup_deadline: Instant,
     ) -> Result<Self, DynamoError> {
-        let pool = GrpcChannelPool::connect("vLLM", endpoint, transport).await?;
+        let pool = timeout_at(
+            startup_deadline,
+            GrpcChannelPool::connect("vLLM", endpoint, transport),
+        )
+        .await
+        .map_err(|_| {
+            dynamo_sidecar_common::connection_timeout(format!(
+                "vLLM gRPC connection pool to {endpoint} exceeded the total startup deadline"
+            ))
+        })??;
         Ok(Self { pool })
     }
 
@@ -27,11 +45,110 @@ impl VllmClient {
         self.pool.len()
     }
 
+    pub(crate) async fn wait_for_services(
+        &self,
+        services: &[&str],
+        startup_deadline: Instant,
+        retry_interval: Duration,
+    ) -> Result<(), DynamoError> {
+        for service in services {
+            self.wait_for_service(service, startup_deadline, retry_interval)
+                .await?;
+        }
+        Ok(())
+    }
+
+    async fn wait_for_service(
+        &self,
+        service: &str,
+        deadline: Instant,
+        retry_interval: Duration,
+    ) -> Result<(), DynamoError> {
+        let started = Instant::now();
+        loop {
+            let mut client = HealthClient::new(self.pool.next_channel());
+            let last_status = match timeout_at(
+                deadline,
+                client.check(HealthCheckRequest {
+                    service: service.to_string(),
+                }),
+            )
+            .await
+            {
+                Ok(Ok(response)) => {
+                    let status = ServingStatus::try_from(response.into_inner().status)
+                        .unwrap_or(ServingStatus::Unknown);
+                    if status == ServingStatus::Serving {
+                        return Ok(());
+                    }
+                    format!("reported {}", status.as_str_name())
+                }
+                Ok(Err(status))
+                    if matches!(
+                        status.code(),
+                        tonic::Code::NotFound | tonic::Code::Unimplemented
+                    ) =>
+                {
+                    return Err(protocol_error(format!(
+                        "{service} is unavailable through the standard gRPC health API: {status}"
+                    )));
+                }
+                Ok(Err(status)) => format!("health check failed: {status}"),
+                Err(_) => "health check exceeded the startup deadline".to_string(),
+            };
+
+            let now = Instant::now();
+            if now >= deadline {
+                return Err(dynamo_sidecar_common::cannot_connect(format!(
+                    "{service} did not become SERVING before the total vLLM startup deadline after {:?}: {last_status}",
+                    started.elapsed()
+                )));
+            }
+            let retry_at = now.checked_add(retry_interval).unwrap_or(deadline);
+            sleep_until(retry_at.min(deadline)).await;
+        }
+    }
+
+    pub(crate) async fn discover(
+        &self,
+        startup_deadline: Instant,
+    ) -> Result<(pb::ModelInfo, pb::ServerInfo), DynamoError> {
+        let channel = self.pool.next_channel();
+        let mut client = pb::control_client::ControlClient::new(channel)
+            .max_encoding_message_size(DEFAULT_MAX_GRPC_MESSAGE_SIZE)
+            .max_decoding_message_size(DEFAULT_MAX_GRPC_MESSAGE_SIZE);
+        let model = timeout_at(
+            startup_deadline,
+            client.get_model_info(pb::GetModelInfoRequest {}),
+        )
+        .await
+        .map_err(|_| {
+            dynamo_sidecar_common::connection_timeout(
+                "GetModelInfo exceeded the total vLLM startup deadline",
+            )
+        })?
+        .map(tonic::Response::into_inner)
+        .map_err(|status| status_to_dynamo("GetModelInfo", status))?;
+        let server = timeout_at(
+            startup_deadline,
+            client.get_server_info(pb::GetServerInfoRequest {}),
+        )
+        .await
+        .map_err(|_| {
+            dynamo_sidecar_common::connection_timeout(
+                "GetServerInfo exceeded the total vLLM startup deadline",
+            )
+        })?
+        .map(tonic::Response::into_inner)
+        .map_err(|status| status_to_dynamo("GetServerInfo", status))?;
+        Ok((model, server))
+    }
+
     pub(crate) async fn generate_stream(
         &self,
         request: pb::GenerateRequest,
     ) -> Result<tonic::Streaming<pb::GenerateResponse>, DynamoError> {
-        let mut client = pb::generate_client::GenerateClient::new(self.pool.next_channel())
+        let mut client = pb::inference_client::InferenceClient::new(self.pool.next_channel())
             .max_encoding_message_size(DEFAULT_MAX_GRPC_MESSAGE_SIZE)
             .max_decoding_message_size(DEFAULT_MAX_GRPC_MESSAGE_SIZE);
         client
@@ -42,6 +159,14 @@ impl VllmClient {
     }
 }
 
+pub(crate) fn startup_deadline(duration: Duration) -> Result<Instant, DynamoError> {
+    Instant::now().checked_add(duration).ok_or_else(|| {
+        invalid_argument(format!(
+            "gRPC startup deadline {duration:?} exceeds the supported monotonic clock range"
+        ))
+    })
+}
+
 pub(crate) fn protocol_error(message: impl Into<String>) -> DynamoError {
     dynamo_sidecar_common::protocol_error("vLLM", message)
 }
diff --git a/lib/sidecar/vllm/src/convert.rs b/lib/sidecar/vllm/src/convert.rs
index 2fceda775cb6..193f5c21c1dc 100644
--- a/lib/sidecar/vllm/src/convert.rs
+++ b/lib/sidecar/vllm/src/convert.rs
@@ -90,6 +90,9 @@ pub(crate) fn build_generate_request(
         kv: Some(kv),
         truncate_prompt_tokens: 0,
         priority,
+        session_id: None,
+        media: Vec::new(),
+        data_parallel_rank: None,
     })
 }
 
@@ -241,6 +244,7 @@ fn build_kv_parameters(
         bypass_prefix_cache,
         cache_salt: cache_salt.unwrap_or_default(),
         kv_transfer_params: kv_transfer_params.map(json_to_struct).transpose()?,
+        ec_transfer_params: None,
     })
 }
 
diff --git a/lib/sidecar/vllm/src/engine.rs b/lib/sidecar/vllm/src/engine.rs
index da2fffe74dec..347e93e09db4 100644
--- a/lib/sidecar/vllm/src/engine.rs
+++ b/lib/sidecar/vllm/src/engine.rs
@@ -9,16 +9,17 @@ use dynamo_backend_common::{
 use dynamo_sidecar_common::{GrpcEndpoint, GrpcTransportConfig};
 use futures::stream::BoxStream;
 use tokio::sync::OnceCell;
+use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 
 use crate::args::Args;
-use crate::client::{self, VllmClient};
+use crate::client::{self, CONTROL_SERVICE, INFERENCE_SERVICE, VllmClient};
 use crate::convert::{ResponseState, build_generate_request};
-use crate::model::ConfiguredModel;
+use crate::model::DiscoveredModel;
 
 pub struct VllmSidecarEngine {
     endpoint: GrpcEndpoint,
-    model: ConfiguredModel,
+    model: DiscoveredModel,
     mode: DisaggregationMode,
     transport: GrpcTransportConfig,
     client: OnceCell<VllmClient>,
@@ -35,7 +36,7 @@ fn cancelled(state: &ResponseState) -> LLMEngineOutput {
 impl VllmSidecarEngine {
     pub(crate) fn new(
         endpoint: GrpcEndpoint,
-        model: ConfiguredModel,
+        model: DiscoveredModel,
         mode: DisaggregationMode,
         transport: GrpcTransportConfig,
     ) -> Self {
@@ -49,6 +50,11 @@ impl VllmSidecarEngine {
         }
     }
 
+    /// Parse arguments and synchronously discover the vLLM model.
+    ///
+    /// Call this before `dynamo_backend_common::run`. Async callers must use
+    /// `spawn_blocking` or a dedicated thread because discovery uses
+    /// `Runtime::block_on`.
     pub fn from_args(argv: Option<Vec<String>>) -> Result<(Self, WorkerConfig), DynamoError> {
         let parsing_process_args = argv.is_none();
         let parsed = match argv {
@@ -73,9 +79,6 @@ impl VllmSidecarEngine {
     }
 
     fn from_parsed(args: Args) -> Result<(Self, WorkerConfig), DynamoError> {
-        if args.model_path.trim().is_empty() {
-            return Err(client::invalid_argument("model-path must not be empty"));
-        }
         if args.sidecar.common.disaggregation_mode.is_encode() {
             return Err(client::invalid_argument(
                 "encode mode is not supported by the vLLM sidecar",
@@ -86,22 +89,24 @@ impl VllmSidecarEngine {
                 "route-to-encoder is not supported by the vLLM sidecar",
             ));
         }
+        if args.sidecar.common.dyn_tool_call_parser.is_some()
+            || args.sidecar.common.dyn_reasoning_parser.is_some()
+        {
+            return Err(client::invalid_argument(
+                "vLLM gRPC does not preserve the request options required by Dynamo tool-call and reasoning parsers",
+            ));
+        }
 
         let endpoint = GrpcEndpoint::parse(&args.vllm_endpoint, "--vllm-endpoint")?;
         let transport = args.sidecar.grpc.config();
-        let model = ConfiguredModel {
-            source: args.model_path,
-        };
+        let bootstrap_deadline = client::startup_deadline(transport.startup_deadline)?;
+        eprintln!(
+            "Discovering vLLM model metadata from {endpoint}; startup deadline: {:?}",
+            transport.startup_deadline
+        );
+        let model = bootstrap_discover(&endpoint, transport, bootstrap_deadline)?;
         let mode = args.sidecar.common.disaggregation_mode;
         let engine = Self::new(endpoint, model.clone(), mode, transport);
-        let (tool_call_parser, reasoning_parser) = if mode.is_prefill() {
-            (None, None)
-        } else {
-            (
-                args.sidecar.common.dyn_tool_call_parser,
-                args.sidecar.common.dyn_reasoning_parser,
-            )
-        };
         let config = WorkerConfig {
             namespace: args.sidecar.common.namespace,
             // Prefill/decode must register under fixed role components so the
@@ -115,9 +120,9 @@ impl VllmSidecarEngine {
             endpoint_types: args.sidecar.common.endpoint_types,
             custom_jinja_template: args.sidecar.common.custom_jinja_template,
             model_name: model.source.clone(),
-            served_model_name: None,
-            tool_call_parser,
-            reasoning_parser,
+            served_model_name: Some(model.served_name.clone()),
+            tool_call_parser: None,
+            reasoning_parser: None,
             exclude_tools_when_tool_choice_none: args
                 .sidecar
                 .common
@@ -146,7 +151,18 @@ impl LLMEngine for VllmSidecarEngine {
             mode = %self.mode,
             "connecting to vLLM gRPC"
         );
-        let client = VllmClient::connect(&self.endpoint, self.transport).await?;
+        let startup_deadline = client::startup_deadline(self.transport.startup_deadline)?;
+        let client = VllmClient::connect(&self.endpoint, self.transport, startup_deadline).await?;
+        client
+            .wait_for_services(
+                &[CONTROL_SERVICE, INFERENCE_SERVICE],
+                startup_deadline,
+                self.transport.retry_interval,
+            )
+            .await?;
+        let (model, server) = client.discover(startup_deadline).await?;
+        let observed = DiscoveredModel::from_proto(model, server)?;
+        self.model.ensure_same_identity(&observed)?;
         let connection_count = client.connection_count();
         self.client
             .set(client)
@@ -154,11 +170,12 @@ impl LLMEngine for VllmSidecarEngine {
         tracing::info!(
             endpoint = %self.endpoint,
             connections = connection_count,
-            configured_model_source = %self.model.source,
+            model = %observed.source,
+            served_model_name = %observed.served_name,
             mode = %self.mode,
-            "vLLM gRPC transport connected"
+            "vLLM gRPC services are ready"
         );
-        Ok(self.model.engine_config())
+        Ok(observed.engine_config())
     }
 
     async fn generate(
@@ -172,7 +189,8 @@ impl LLMEngine for VllmSidecarEngine {
             .ok_or_else(|| client::engine_shutdown("vLLM sidecar is not started"))?;
         let request_id = ctx.id().to_string();
         let mut state = ResponseState::new(&request, self.mode);
-        let proto_request = build_generate_request(request, request_id, self.mode)?;
+        let mut proto_request = build_generate_request(request, request_id, self.mode)?;
+        proto_request.model.clone_from(&self.model.served_name);
         let stopped_ctx = ctx.inner_arc();
         let shutdown = self.cancel.clone();
         let mut cancellation = Box::pin(async move {
@@ -237,3 +255,30 @@ impl LLMEngine for VllmSidecarEngine {
         Ok(())
     }
 }
+
+fn bootstrap_discover(
+    endpoint: &GrpcEndpoint,
+    transport: GrpcTransportConfig,
+    startup_deadline: Instant,
+) -> Result<DiscoveredModel, DynamoError> {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .map_err(|error| client::engine_shutdown(format!("bootstrap runtime: {error}")))?;
+    runtime.block_on(async {
+        let bootstrap_transport = GrpcTransportConfig {
+            connections: std::num::NonZeroUsize::MIN,
+            ..transport
+        };
+        let client = VllmClient::connect(endpoint, bootstrap_transport, startup_deadline).await?;
+        client
+            .wait_for_services(
+                &[CONTROL_SERVICE],
+                startup_deadline,
+                transport.retry_interval,
+            )
+            .await?;
+        let (model, server) = client.discover(startup_deadline).await?;
+        DiscoveredModel::from_proto(model, server)
+    })
+}
diff --git a/lib/sidecar/vllm/src/model.rs b/lib/sidecar/vllm/src/model.rs
index 7d242fb17e9d..b40c3191191c 100644
--- a/lib/sidecar/vllm/src/model.rs
+++ b/lib/sidecar/vllm/src/model.rs
@@ -1,20 +1,109 @@
 // SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-use dynamo_backend_common::{EngineConfig, LlmRegistration};
+use dynamo_backend_common::{DynamoError, EngineConfig, LlmRegistration};
+
+use crate::client;
+use crate::proto as pb;
+
+const SUPPORTED_API_VERSION: &str = "vllm";
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+struct ModelIdentity {
+    source: String,
+    served_name: String,
+    aliases: Vec<String>,
+    reasoning_parser: Option<String>,
+    tool_call_parser: Option<String>,
+}
 
 #[derive(Clone, Debug)]
-pub(crate) struct ConfiguredModel {
+pub(crate) struct DiscoveredModel {
     pub source: String,
+    pub served_name: String,
+    identity: ModelIdentity,
+    server: pb::ServerInfo,
 }
 
-impl ConfiguredModel {
+impl DiscoveredModel {
+    pub(crate) fn from_proto(
+        model: pb::ModelInfo,
+        server: pb::ServerInfo,
+    ) -> Result<Self, DynamoError> {
+        if server.api_version != SUPPORTED_API_VERSION {
+            return Err(client::protocol_error(format!(
+                "unsupported Control API version `{}`; expected `{SUPPORTED_API_VERSION}`",
+                server.api_version
+            )));
+        }
+        let source = required("model_id", model.model_id)?;
+        let served_name = required("served_model_name", model.served_model_name)?;
+        if !model.supports_token_ids_input {
+            return Err(client::protocol_error(
+                "the discovered model does not support token-ID input",
+            ));
+        }
+        let reasoning_parser = nonempty(model.reasoning_parser);
+        let tool_call_parser = nonempty(model.tool_call_parser);
+        let identity = ModelIdentity {
+            source: source.clone(),
+            served_name: served_name.clone(),
+            aliases: model.served_model_aliases,
+            reasoning_parser: reasoning_parser.clone(),
+            tool_call_parser: tool_call_parser.clone(),
+        };
+        Ok(Self {
+            source,
+            served_name,
+            identity,
+            server,
+        })
+    }
+
+    pub(crate) fn ensure_same_identity(&self, observed: &Self) -> Result<(), DynamoError> {
+        if self.identity != observed.identity {
+            return Err(client::protocol_error(format!(
+                "model identity changed between bootstrap and startup: expected {:?}, observed {:?}",
+                self.identity, observed.identity
+            )));
+        }
+        Ok(())
+    }
+
     pub(crate) fn engine_config(&self) -> EngineConfig {
         EngineConfig {
             model: self.source.clone(),
-            served_model_name: None,
+            served_model_name: Some(self.served_name.clone()),
+            model_aliases: self.identity.aliases.clone(),
             runtime_data: Default::default(),
-            llm: Some(LlmRegistration::default()),
+            llm: Some(LlmRegistration {
+                context_length: nonzero(self.server.max_model_len),
+                kv_cache_block_size: nonzero(self.server.kv_block_size),
+                total_kv_blocks: nonzero(self.server.total_kv_blocks),
+                max_num_seqs: nonzero(self.server.max_running_requests),
+                max_num_batched_tokens: nonzero(self.server.max_batched_tokens),
+                ..Default::default()
+            }),
         }
     }
 }
+
+fn required(field: &str, value: String) -> Result<String, DynamoError> {
+    if value.trim().is_empty() {
+        return Err(client::protocol_error(format!(
+            "Control returned an empty {field}"
+        )));
+    }
+    Ok(value)
+}
+
+fn nonempty(value: String) -> Option<String> {
+    (!value.trim().is_empty()).then_some(value)
+}
+
+fn nonzero<T>(value: T) -> Option<T>
+where
+    T: Default + PartialEq,
+{
+    (value != T::default()).then_some(value)
+}
diff --git a/lib/sidecar/vllm/src/tests.rs b/lib/sidecar/vllm/src/tests.rs
index ee05667f7982..05043e15cc10 100644
--- a/lib/sidecar/vllm/src/tests.rs
+++ b/lib/sidecar/vllm/src/tests.rs
@@ -19,18 +19,20 @@ use tokio::net::TcpListener;
 use tokio::sync::{Mutex, Notify, oneshot};
 use tokio_stream::wrappers::TcpListenerStream;
 use tonic::{Request, Response, Status};
+use tonic_health::ServingStatus as HealthServingStatus;
 
-use crate::client::VllmClient;
+use crate::client::{CONTROL_SERVICE, INFERENCE_SERVICE, VllmClient};
 use crate::convert::{ResponseState, build_generate_request};
 use crate::engine::VllmSidecarEngine;
 use crate::json::{json_to_struct, struct_to_json};
-use crate::model::ConfiguredModel;
+use crate::model::DiscoveredModel;
 use crate::proto as pb;
 
 #[derive(Clone, Default)]
-struct FakeGenerate {
+struct FakeVllm {
     requests: Arc<Mutex<Vec<pb::GenerateRequest>>>,
     peers: Arc<Mutex<Vec<SocketAddr>>>,
+    model_info_override: Arc<Mutex<Option<pb::ModelInfo>>>,
     reject: Arc<AtomicBool>,
     hang: Arc<AtomicBool>,
     hang_before_headers: Arc<AtomicBool>,
@@ -48,7 +50,7 @@ impl Drop for DropSignal {
 }
 
 #[tonic::async_trait]
-impl pb::generate_server::Generate for FakeGenerate {
+impl pb::inference_server::Inference for FakeVllm {
     type GenerateStreamStream =
         Pin<Box<dyn Stream<Item = Result<pb::GenerateResponse, Status>> + Send>>;
 
@@ -161,6 +163,79 @@ impl pb::generate_server::Generate for FakeGenerate {
     }
 }
 
+#[tonic::async_trait]
+impl pb::control_server::Control for FakeVllm {
+    async fn get_server_info(
+        &self,
+        _request: Request<pb::GetServerInfoRequest>,
+    ) -> Result<Response<pb::ServerInfo>, Status> {
+        Ok(Response::new(server_info()))
+    }
+
+    async fn get_model_info(
+        &self,
+        _request: Request<pb::GetModelInfoRequest>,
+    ) -> Result<Response<pb::ModelInfo>, Status> {
+        let model = self
+            .model_info_override
+            .lock()
+            .await
+            .clone()
+            .unwrap_or_else(model_info);
+        Ok(Response::new(model))
+    }
+
+    async fn abort(
+        &self,
+        _request: Request<pb::AbortRequest>,
+    ) -> Result<Response<pb::AbortResponse>, Status> {
+        Ok(Response::new(pb::AbortResponse {}))
+    }
+
+    async fn get_kv_event_sources(
+        &self,
+        _request: Request<pb::GetKvEventSourcesRequest>,
+    ) -> Result<Response<pb::GetKvEventSourcesResponse>, Status> {
+        Ok(Response::new(pb::GetKvEventSourcesResponse {
+            sources: Vec::new(),
+        }))
+    }
+}
+
+fn model_info() -> pb::ModelInfo {
+    pb::ModelInfo {
+        model_id: "model-source".to_string(),
+        served_model_name: "served-model".to_string(),
+        served_model_aliases: vec!["model-alias".to_string()],
+        supports_text_input: true,
+        supports_token_ids_input: true,
+        supports_multimodal: false,
+        reasoning_parser: "deepseek_r1".to_string(),
+        tool_call_parser: "hermes".to_string(),
+    }
+}
+
+fn server_info() -> pb::ServerInfo {
+    pb::ServerInfo {
+        engine_version: "test-vllm".to_string(),
+        api_version: "vllm".to_string(),
+        instance_id: "test-instance".to_string(),
+        parallelism: Some(pb::ParallelismInfo {
+            tensor_parallel_size: 2,
+            pipeline_parallel_size: 1,
+            data_parallel_size: 4,
+            data_parallel_rank: 2,
+            decode_context_parallel_size: 1,
+        }),
+        max_model_len: 8192,
+        kv_block_size: 16,
+        total_kv_blocks: 4096,
+        max_running_requests: 128,
+        max_batched_tokens: 2048,
+        supports_explicit_data_parallel_rank: false,
+    }
+}
+
 fn sequence_response(
     terminal: bool,
     logprobs: bool,
@@ -189,6 +264,7 @@ fn sequence_response(
                 finish_reason: pb::finish_info::FinishReason::Stop as i32,
                 stop_reason: Some(pb::finish_info::StopReason::StopTokenId(2)),
                 kv_transfer_params,
+                ec_transfer_params: None,
             }),
         }),
     }
@@ -313,23 +389,33 @@ fn oversized_logprob_counts_are_rejected() {
 
 struct FakeServer {
     endpoint: String,
-    service: FakeGenerate,
+    service: FakeVllm,
     shutdown: Option<oneshot::Sender<()>>,
 }
 
 impl FakeServer {
-    async fn start(service: FakeGenerate) -> Self {
+    async fn start(service: FakeVllm) -> Self {
         let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
         let address = listener.local_addr().expect("address");
         let (shutdown, shutdown_rx) = oneshot::channel();
-        let server_service = service.clone();
+        let inference_service = service.clone();
+        let control_service = service.clone();
+        let (health, health_service) = tonic_health::server::health_reporter();
+        health
+            .set_service_status(CONTROL_SERVICE, HealthServingStatus::Serving)
+            .await;
+        health
+            .set_service_status(INFERENCE_SERVICE, HealthServingStatus::Serving)
+            .await;
         tokio::spawn(async move {
             tonic::transport::Server::builder()
                 .add_service(
-                    pb::generate_server::GenerateServer::new(server_service)
+                    pb::inference_server::InferenceServer::new(inference_service)
                         .max_encoding_message_size(64 * 1024 * 1024)
                         .max_decoding_message_size(64 * 1024 * 1024),
                 )
+                .add_service(pb::control_server::ControlServer::new(control_service))
+                .add_service(health_service)
                 .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async {
                     let _ = shutdown_rx.await;
                 })
@@ -397,19 +483,38 @@ fn request() -> PreprocessedRequest {
 }
 
 fn engine(endpoint: &str, mode: DisaggregationMode, connections: usize) -> VllmSidecarEngine {
+    let transport = GrpcTransportConfig {
+        connections: NonZeroUsize::new(connections).expect("non-zero connection count"),
+        ..Default::default()
+    };
     VllmSidecarEngine::new(
         GrpcEndpoint::parse(endpoint, "--vllm-endpoint").expect("valid test endpoint"),
-        ConfiguredModel {
-            source: "model-source".to_string(),
-        },
+        DiscoveredModel::from_proto(model_info(), server_info()).expect("valid discovery"),
         mode,
-        GrpcTransportConfig {
-            connections: NonZeroUsize::new(connections).expect("non-zero connection count"),
-            ..Default::default()
-        },
+        transport,
     )
 }
 
+async fn engine_from_args(
+    endpoint: &str,
+) -> (VllmSidecarEngine, dynamo_backend_common::WorkerConfig) {
+    let argv = vec![
+        "dynamo-vllm-sidecar".to_string(),
+        "--vllm-endpoint".to_string(),
+        endpoint.to_string(),
+        "--grpc-connections".to_string(),
+        "2".to_string(),
+        "--grpc-startup-deadline-secs".to_string(),
+        "5".to_string(),
+        "--grpc-connect-attempt-timeout-secs".to_string(),
+        "1".to_string(),
+    ];
+    tokio::task::spawn_blocking(move || VllmSidecarEngine::from_args(Some(argv)))
+        .await
+        .expect("bootstrap task")
+        .expect("bootstrap discovery")
+}
+
 async fn collect(
     engine: &VllmSidecarEngine,
     request: PreprocessedRequest,
@@ -424,13 +529,65 @@ async fn collect(
         .await
 }
 
+#[test]
+fn discovery_rejects_incompatible_model_metadata() {
+    let mut unsupported_api = server_info();
+    unsupported_api.api_version = "unsupported".to_string();
+
+    let mut missing_model_id = model_info();
+    missing_model_id.model_id.clear();
+
+    let mut missing_served_name = model_info();
+    missing_served_name.served_model_name.clear();
+
+    let mut unsupported_input = model_info();
+    unsupported_input.supports_token_ids_input = false;
+
+    for (case, model, server) in [
+        ("unsupported API", model_info(), unsupported_api),
+        ("missing model ID", missing_model_id, server_info()),
+        ("missing served name", missing_served_name, server_info()),
+        ("unsupported input", unsupported_input, server_info()),
+    ] {
+        assert!(
+            DiscoveredModel::from_proto(model, server).is_err(),
+            "{case} metadata should be rejected"
+        );
+    }
+}
+
+#[tokio::test]
+async fn startup_rejects_model_identity_change_after_bootstrap() {
+    let server = FakeServer::start(FakeVllm::default()).await;
+    let (engine, _) = engine_from_args(&server.endpoint).await;
+
+    let mut changed = model_info();
+    changed.served_model_name = "changed-served-model".to_string();
+    *server.service.model_info_override.lock().await = Some(changed);
+
+    assert!(engine.start(0).await.is_err());
+}
+
 #[tokio::test]
 async fn aggregated_generation_converts_request_stream_and_usage() {
-    let server = FakeServer::start(FakeGenerate::default()).await;
-    let engine = engine(&server.endpoint, DisaggregationMode::Aggregated, 2);
+    let server = FakeServer::start(FakeVllm::default()).await;
+    let (engine, worker) = engine_from_args(&server.endpoint).await;
+    assert_eq!(worker.model_name, "model-source");
+    assert_eq!(worker.served_model_name.as_deref(), Some("served-model"));
+    assert!(worker.reasoning_parser.is_none());
+    assert!(worker.tool_call_parser.is_none());
     let config = engine.start(0).await.expect("start");
     assert_eq!(config.model, "model-source");
-    assert_eq!(config.served_model_name, None);
+    assert_eq!(config.served_model_name.as_deref(), Some("served-model"));
+    assert_eq!(config.model_aliases, ["model-alias"]);
+    let registration = config.llm.expect("LLM registration");
+    assert_eq!(registration.context_length, Some(8192));
+    assert_eq!(registration.kv_cache_block_size, Some(16));
+    assert_eq!(registration.total_kv_blocks, Some(4096));
+    assert_eq!(registration.max_num_seqs, Some(128));
+    assert_eq!(registration.max_num_batched_tokens, Some(2048));
+    assert_eq!(registration.data_parallel_size, None);
+    assert_eq!(registration.data_parallel_start_rank, None);
 
     let outputs = collect(&engine, request()).await;
     assert_eq!(outputs.len(), 1);
@@ -446,7 +603,7 @@ async fn aggregated_generation_converts_request_stream_and_usage() {
 
     let requests = server.service.requests.lock().await;
     let sent = requests.first().expect("recorded request");
-    assert!(sent.model.is_empty());
+    assert_eq!(sent.model, "served-model");
     assert_eq!(sent.priority, 0);
     let sampling = sent.sampling.as_ref().unwrap();
     assert_eq!(
@@ -483,7 +640,7 @@ async fn aggregated_generation_converts_request_stream_and_usage() {
 
 #[tokio::test]
 async fn grpc_request_errors_are_propagated() {
-    let service = FakeGenerate::default();
+    let service = FakeVllm::default();
     service.reject.store(true, Ordering::SeqCst);
     let server = FakeServer::start(service).await;
     let engine = engine(&server.endpoint, DisaggregationMode::Aggregated, 1);
@@ -499,7 +656,7 @@ async fn grpc_request_errors_are_propagated() {
 
 #[tokio::test]
 async fn prefill_decode_handoff_is_opaque_and_repeatable() {
-    let server = FakeServer::start(FakeGenerate::default()).await;
+    let server = FakeServer::start(FakeVllm::default()).await;
     let prefill = engine(&server.endpoint, DisaggregationMode::Prefill, 1);
     let decode = engine(&server.endpoint, DisaggregationMode::Decode, 1);
     prefill.start(0).await.expect("start prefill");
@@ -534,40 +691,43 @@ async fn prefill_decode_handoff_is_opaque_and_repeatable() {
     }
 }
 
-#[test]
-fn component_honors_config_for_aggregated_but_fixes_disagg_roles() {
-    let component = |extra: &[&str]| {
+#[tokio::test]
+async fn component_honors_config_for_aggregated_but_fixes_disagg_roles() {
+    let server = FakeServer::start(FakeVllm::default()).await;
+    for (extra, expected) in [
+        (Vec::<&str>::new(), "custom"),
+        (vec!["--disaggregation-mode", "prefill"], "prefill"),
+        (vec!["--disaggregation-mode", "decode"], "backend"),
+    ] {
         let mut argv = vec![
-            "dynamo-vllm-sidecar",
-            "--vllm-endpoint",
-            "127.0.0.1:50051",
-            "--model-path",
-            "test-model",
-            "--component",
-            "custom",
+            "dynamo-vllm-sidecar".to_string(),
+            "--vllm-endpoint".to_string(),
+            server.endpoint.clone(),
+            "--component".to_string(),
+            "custom".to_string(),
         ];
-        argv.extend_from_slice(extra);
-        VllmSidecarEngine::from_args(Some(argv.iter().map(|s| s.to_string()).collect()))
-            .expect("from_args")
-            .1
-            .component
-    };
-    // Aggregated keeps the operator-configured component.
-    assert_eq!(component(&[]), "custom");
-    // Disaggregated roles override to fixed names so the frontend can route.
-    assert_eq!(component(&["--disaggregation-mode", "prefill"]), "prefill");
-    assert_eq!(component(&["--disaggregation-mode", "decode"]), "backend");
+        argv.extend(extra.into_iter().map(str::to_string));
+        let component =
+            tokio::task::spawn_blocking(move || VllmSidecarEngine::from_args(Some(argv)))
+                .await
+                .expect("bootstrap task")
+                .expect("from_args")
+                .1
+                .component;
+        assert_eq!(component, expected);
+    }
 }
 
 #[tokio::test]
 async fn pool_uses_each_configured_connection() {
-    let server = FakeServer::start(FakeGenerate::default()).await;
+    let server = FakeServer::start(FakeVllm::default()).await;
     let transport = GrpcTransportConfig {
         connections: NonZeroUsize::new(2).unwrap(),
         ..Default::default()
     };
     let endpoint = GrpcEndpoint::parse(&server.endpoint, "--vllm-endpoint").unwrap();
-    let client = VllmClient::connect(&endpoint, transport)
+    let deadline = crate::client::startup_deadline(transport.startup_deadline).unwrap();
+    let client = VllmClient::connect(&endpoint, transport, deadline)
         .await
         .expect("connect pool");
     assert_eq!(client.connection_count(), 2);
@@ -597,7 +757,7 @@ async fn pool_uses_each_configured_connection() {
 
 #[tokio::test]
 async fn cancellation_drops_the_remote_stream() {
-    let service = FakeGenerate::default();
+    let service = FakeVllm::default();
     service.hang.store(true, Ordering::SeqCst);
     let server = FakeServer::start(service).await;
     let engine = engine(&server.endpoint, DisaggregationMode::Aggregated, 1);
@@ -626,7 +786,7 @@ async fn cancellation_drops_the_remote_stream() {
 
 #[tokio::test]
 async fn cancellation_interrupts_pending_response_headers() {
-    let service = FakeGenerate::default();
+    let service = FakeVllm::default();
     service.hang_before_headers.store(true, Ordering::SeqCst);
     let server = FakeServer::start(service).await;
     let engine = engine(&server.endpoint, DisaggregationMode::Aggregated, 1);
@@ -657,7 +817,7 @@ async fn cancellation_interrupts_pending_response_headers() {
 
 #[tokio::test]
 async fn unsupported_features_fail_before_rpc_submission() {
-    let server = FakeServer::start(FakeGenerate::default()).await;
+    let server = FakeServer::start(FakeVllm::default()).await;
     let engine = engine(&server.endpoint, DisaggregationMode::Aggregated, 1);
     engine.start(0).await.expect("start");
 
diff --git a/lib/sidecar/vllm/tests/executable.rs b/lib/sidecar/vllm/tests/executable.rs
index 66677148a508..2e1320afb4fd 100644
--- a/lib/sidecar/vllm/tests/executable.rs
+++ b/lib/sidecar/vllm/tests/executable.rs
@@ -19,7 +19,6 @@ fn executable_exposes_native_grpc_configuration() {
     for flag in [
         "--vllm-endpoint",
         "--grpc-connections",
-        "--model-path",
         "--disaggregation-mode",
         "--grpc-connect-attempt-timeout-secs",
         "--grpc-retry-interval-secs",
@@ -27,7 +26,6 @@ fn executable_exposes_native_grpc_configuration() {
     ] {
         assert!(stdout.contains(flag), "missing {flag} in help output");
     }
-
     for env in [
         "DYN_SIDECAR_GRPC_CONNECTIONS",
         "DYN_SIDECAR_GRPC_CONNECT_ATTEMPT_TIMEOUT_SECS",