[Rust Frontend] Extend the existing gRPC service protocol

rust/Cargo.toml

@@ -111,6 +111,7 @@ tokio-stream = "0.1"
 tokio-util = { version = "0.7.18", features = ["rt"] }
 tonic = "0.14.5"
 tonic-build = "0.14.5"
+tonic-health = "0.14.5"
 tonic-prost = "0.14.5"
 tonic-prost-build = "0.14.5"
 tool-parser = "1.2.0"

rust/proto/vllm_grpc.proto

@@ -11,6 +11,19 @@ service Generate {
   rpc GenerateStream (GenerateRequest) returns (stream GenerateResponse) {}
 }
 
+service Control {
+  rpc GetServerInfo (GetServerInfoRequest) returns (ServerInfo) {}
+  rpc GetModelInfo (GetModelInfoRequest) returns (ModelInfo) {}
+  rpc Abort (AbortRequest) returns (AbortResponse) {}
+  rpc Drain (DrainRequest) returns (DrainResponse) {}
+
+  rpc LoadLora (LoadLoraRequest) returns (LoadLoraResponse) {}
+  rpc UnloadLora (UnloadLoraRequest) returns (UnloadLoraResponse) {}
+  rpc ListLoras (ListLorasRequest) returns (ListLorasResponse) {}
+
+  rpc GetKvEventSources (GetKvEventSourcesRequest) returns (GetKvEventSourcesResponse) {}
+}
+
 // ======================================================================================
 // Generate Request
 // ======================================================================================
@@ -42,6 +55,11 @@ message GenerateRequest {
   uint32 truncate_prompt_tokens = 11;
 
   int32 priority = 12;
+
+  // Multimodal inputs aligned with placeholder markers in token_ids.
+  repeated MediaItem media = 13;
+  // Loaded LoRA name; empty selects the base model.
+  string lora_name = 14;
 }
 
 message RandomSampling {
@@ -194,3 +212,134 @@ 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;
+    string data_uri = 3;
+    bytes raw_bytes = 4;
+  }
+  string mime_type = 5;
+  string uuid = 6;
+}
+
+// ======================================================================================
+// Discovery and lifecycle
+// ======================================================================================
+
+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;
+  uint32 max_loras = 10;
+}
+
+message ParallelismInfo {
+  uint32 tensor_parallel_size = 1;
+  uint32 pipeline_parallel_size = 2;
+  uint32 data_parallel_size = 3;
+  uint32 data_parallel_rank = 4;
+  uint32 data_parallel_start_rank = 5;
+  uint32 decode_context_parallel_size = 6;
+}
+
+message GetModelInfoRequest {}
+
+message ModelInfo {
+  string model_id = 1;
+  string served_model_name = 2;
+  repeated string served_model_aliases = 3;
+  repeated string tokenizer_modes = 4;
+
+  bool supports_text_input = 20;
+  bool supports_token_ids_input = 21;
+  bool supports_lora = 22;
+  bool supports_multimodal = 23;
+  string reasoning_parser = 24;
+  string tool_call_parser = 25;
+}
+
+message AbortRequest {
+  repeated string request_ids = 1;
+}
+
+message AbortResponse {}
+
+message DrainRequest {}
+
+message DrainResponse {
+  DrainState state = 1;
+  uint32 in_flight_requests = 2;
+  string message = 3;
+}
+
+enum DrainState {
+  DRAIN_STATE_UNSPECIFIED = 0;
+  DRAIN_STATE_IN_PROGRESS = 1;
+  DRAIN_STATE_COMPLETE = 2;
+}
+
+// ======================================================================================
+// LoRA lifecycle
+// ======================================================================================
+
+message LoraAdapter {
+  int64 lora_id = 1;
+  string lora_name = 2;
+  string source_path = 3;
+}
+
+message LoadLoraRequest { LoraAdapter adapter = 1; }
+message LoadLoraResponse {
+  LoraAdapter adapter = 1;
+  bool already_loaded = 2;
+}
+message UnloadLoraRequest { string lora_name = 1; }
+message UnloadLoraResponse { LoraAdapter adapter = 1; }
+message ListLorasRequest {}
+message ListLorasResponse { repeated LoraAdapter adapters = 1; }
+
+// ======================================================================================
+// KV discovery
+// ======================================================================================
+
+message GetKvEventSourcesRequest {}
+message GetKvEventSourcesResponse { repeated KvEventSource sources = 1; }
+
+message KvEventEndpoint {
+  string host = 1;
+  uint32 port = 2;
+  string protocol = 3;
+}
+
+message KvEventSource {
+  string transport = 1;
+  KvEventEndpoint endpoint_addr = 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;
+}

rust/src/engine-core-client/src/client.rs

@@ -457,6 +457,12 @@ impl EngineCoreClient {
         self.inner.is_healthy()
     }
 
+    /// Subscribe to engine health changes. The current value is `true` while
+    /// the client is healthy and changes permanently to `false` on failure.
+    pub fn subscribe_health(&self) -> tokio::sync::watch::Receiver<bool> {
+        self.inner.subscribe_health()
+    }
+
     /// Return the first persistent health error observed by the client, if any.
     pub fn health_error(&self) -> Option<Arc<Error>> {
         self.inner.health_error()

rust/src/engine-core-client/src/client/imp.rs

@@ -6,7 +6,7 @@ use arc_swap::ArcSwapOption;
 use parking_lot::Mutex;
 use thiserror_ext::AsReport as _;
 use tokio::runtime::Handle;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, watch};
 use tracing::{debug, info, trace, warn};
 use vllm_metrics::METRICS;
 use zeromq::RouterSendHalf;
@@ -33,6 +33,7 @@ pub(crate) struct ClientInner {
     request_reg: Mutex<RequestRegistry>,
     utility_reg: Mutex<UtilityRegistry>,
     health_error: ArcSwapOption<Error>,
+    health_tx: watch::Sender<bool>,
 }
 
 impl ClientInner {
@@ -54,6 +55,7 @@ impl ClientInner {
             request_reg: Mutex::new(RequestRegistry::new(engines)),
             utility_reg: Mutex::new(UtilityRegistry::default()),
             health_error: ArcSwapOption::empty(),
+            health_tx: watch::channel(true).0,
         }
     }
 
@@ -188,6 +190,12 @@ impl ClientInner {
         self.health_error.load().is_none()
     }
 
+    /// Subscribe to engine health changes. The current value is `true` while
+    /// the client is healthy and changes permanently to `false` on failure.
+    pub fn subscribe_health(&self) -> watch::Receiver<bool> {
+        self.health_tx.subscribe()
+    }
+
     /// Resolve one utility output to the waiting caller. Returns `true` if a
     /// waiting caller existed.
     pub fn resolve_utility_output(&self, output: UtilityOutput) -> bool {
@@ -267,14 +275,23 @@ impl ClientInner {
     /// recorded for this client. Later failures do not overwrite the first
     /// one so `/health` and post-close callers observe a stable cause.
     fn record_health_error(&self, error: Arc<Error>) -> Arc<Error> {
-        if let Some(existing) = self.health_error.load_full() {
-            return existing;
-        }
-        self.health_error
-            .rcu(|current| current.clone().unwrap_or_else(|| error.clone()));
-        self.health_error
-            .load_full()
-            .expect("health error must be recorded before registries close")
+        let persistent_error = if let Some(existing) = self.health_error.load_full() {
+            existing
+        } else {
+            self.health_error
+                .rcu(|current| current.clone().unwrap_or_else(|| error.clone()));
+            self.health_error
+                .load_full()
+                .expect("health error must be recorded before registries close")
+        };
+        self.health_tx.send_if_modified(|healthy| {
+            if !*healthy {
+                return false;
+            }
+            *healthy = false;
+            true
+        });
+        persistent_error
     }
 
     /// Assert there is a recorded health error and return a `Shared` variant
@@ -458,13 +475,18 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn close_registries_records_first_health_error_only() {
         let inner = test_inner().await;
+        let mut health = inner.subscribe_health();
+        assert!(*health.borrow());
 
         inner.close_registries(Arc::new(Error::EngineCoreDead));
+        health.changed().await.expect("health sender remains open");
         assert!(!inner.is_healthy());
+        assert!(!*health.borrow());
         assert!(matches!(
             inner.health_error().as_deref(),
             Some(Error::EngineCoreDead)
         ));
+        assert!(!*inner.subscribe_health().borrow());
 
         inner.close_registries(Arc::new(client_closed!("shutdown")));
         assert!(matches!(

rust/src/server/Cargo.toml

@@ -35,6 +35,7 @@ tokio-openssl.workspace = true
 tokio-stream.workspace = true
 tokio-util.workspace = true
 tonic.workspace = true
+tonic-health.workspace = true
 tonic-prost.workspace = true
 tower.workspace = true
 tower-http.workspace = true