feat(grpc): add RL lifecycle control

docs/training/weight_transfer/README.md

@@ -63,6 +63,8 @@ When running vLLM as an HTTP server, the following endpoints are available for w
 !!! note
     The HTTP weight transfer endpoints require `VLLM_SERVER_DEV_MODE=1` to be set.
 
+The Rust frontend's optional gRPC `Control` service exposes the same pause, sleep, weight-transfer, and weight-version lifecycle for trusted sidecars. The `ServerInfo.rl_capabilities` response reports whether weight transfer and sleep mode were configured. Backend-specific `init_info` and `update_info` remain JSON metadata; model tensors continue to move over the configured NCCL, IPC, or sparse-NCCL transport.
+
 ## Trainer-Side API
 
 Both backends provide static methods that the trainer calls to send weights. The general pattern is:

docs/usage/security.md

@@ -344,7 +344,7 @@ vLLM supports loading out-of-tree HTTP routes via the `vllm.endpoint_plugins` en
 
 ## gRPC Interface
 
-vLLM provides an optional gRPC Generate service on a separate TCP port, enabled via the `--grpc-port` flag. When not specified, no gRPC server is started. The gRPC listener binds to the same host address as the HTTP server.
+vLLM provides optional gRPC `Inference` and `Control` services on a separate TCP port, enabled via the `--grpc-port` flag. When not specified, no gRPC server is started. The gRPC listener binds to the same host address as the HTTP server.
 
 **Warning:** The gRPC interface is **insecure by default** — it does not implement authentication, authorization, or encryption. It should be considered a private, internal interface intended for use only between co-located services within a trusted network. Do not expose the gRPC port to the public internet or untrusted clients. If you enable the gRPC interface, protect it via network-level access controls such as firewall rules, network segmentation, or deployment on an isolated private network.
 
@@ -353,8 +353,9 @@ vLLM provides an optional gRPC Generate service on a separate TCP port, enabled
 An attacker who can reach the gRPC port can:
 
 1. **Run arbitrary inference** via the `Generate` and `GenerateStream` RPCs without any credentials
-2. **Consume GPU and compute resources** by submitting unbounded generation requests
-3. **Cause Denial of Service** by exploiting bugs in the gRPC interface that can crash vLLM.
+2. **Mutate engine state** by pausing generation, sleeping the engine, or initiating configured RL weight updates through the `Control` service
+3. **Consume GPU and compute resources** by submitting unbounded generation requests
+4. **Cause Denial of Service** by exploiting bugs in the gRPC interface that can crash vLLM.
 
 ### Recommendations
 

rust/proto/control.proto

@@ -9,6 +9,21 @@ service Control {
   rpc GetModelInfo (GetModelInfoRequest) returns (ModelInfo) {}
   rpc Abort (AbortRequest) returns (AbortResponse) {}
   rpc GetKvEventSources (GetKvEventSourcesRequest) returns (GetKvEventSourcesResponse) {}
+
+  // Reinforcement-learning lifecycle and weight updates.
+  rpc PauseGeneration (PauseGenerationRequest) returns (PauseGenerationResponse) {}
+  rpc ResumeGeneration (ResumeGenerationRequest) returns (ResumeGenerationResponse) {}
+  rpc IsPaused (IsPausedRequest) returns (IsPausedResponse) {}
+  rpc Sleep (SleepRequest) returns (SleepResponse) {}
+  rpc WakeUp (WakeUpRequest) returns (WakeUpResponse) {}
+  rpc IsSleeping (IsSleepingRequest) returns (IsSleepingResponse) {}
+  rpc InitWeightTransferEngine (InitWeightTransferEngineRequest) returns (InitWeightTransferEngineResponse) {}
+  rpc StartWeightUpdate (StartWeightUpdateRequest) returns (StartWeightUpdateResponse) {}
+  rpc StartDraftWeightUpdate (StartDraftWeightUpdateRequest) returns (StartDraftWeightUpdateResponse) {}
+  rpc UpdateWeights (UpdateWeightsRequest) returns (UpdateWeightsResponse) {}
+  rpc FinishWeightUpdate (FinishWeightUpdateRequest) returns (FinishWeightUpdateResponse) {}
+  rpc UpdateWeightVersion (UpdateWeightVersionRequest) returns (UpdateWeightVersionResponse) {}
+  rpc GetWeightVersion (GetWeightVersionRequest) returns (GetWeightVersionResponse) {}
 }
 
 message GetServerInfoRequest {}
@@ -23,6 +38,14 @@ message ServerInfo {
   uint64 total_kv_blocks = 7;
   uint64 max_running_requests = 8;
   uint64 max_batched_tokens = 9;
+  RlCapabilities rl_capabilities = 11;
+}
+
+message RlCapabilities {
+  bool weight_transfer_enabled = 1;
+  string weight_transfer_backend = 2;
+  bool sleep_mode_enabled = 3;
+  bool draft_weight_updates_enabled = 4;
 }
 
 message ParallelismInfo {
@@ -53,6 +76,64 @@ message AbortRequest {
 
 message AbortResponse {}
 
+// ======================================================================================
+// Reinforcement-learning control
+// ======================================================================================
+
+enum PauseMode {
+  PAUSE_MODE_UNSPECIFIED = 0;
+  PAUSE_MODE_ABORT = 1;
+  PAUSE_MODE_WAIT = 2;
+  PAUSE_MODE_KEEP = 3;
+}
+
+message PauseGenerationRequest {
+  PauseMode mode = 1;
+  optional bool clear_cache = 2;
+}
+message PauseGenerationResponse {}
+
+message ResumeGenerationRequest {}
+message ResumeGenerationResponse {}
+
+message IsPausedRequest {}
+message IsPausedResponse { bool paused = 1; }
+
+message SleepRequest {
+  optional uint32 level = 1;
+  PauseMode mode = 2;
+}
+message SleepResponse {}
+
+message WakeUpRequest { repeated string tags = 1; }
+message WakeUpResponse {}
+
+message IsSleepingRequest {}
+message IsSleepingResponse { bool sleeping = 1; }
+
+// The payloads are backend-specific JSON objects. Tensor data remains on the
+// configured NCCL, IPC, or sparse-NCCL transport rather than crossing gRPC.
+message InitWeightTransferEngineRequest { bytes init_info_json = 1; }
+message InitWeightTransferEngineResponse {}
+
+message StartWeightUpdateRequest {}
+message StartWeightUpdateResponse {}
+
+message StartDraftWeightUpdateRequest {}
+message StartDraftWeightUpdateResponse {}
+
+message UpdateWeightsRequest { bytes update_info_json = 1; }
+message UpdateWeightsResponse {}
+
+message FinishWeightUpdateRequest { optional string weight_version = 1; }
+message FinishWeightUpdateResponse {}
+
+message UpdateWeightVersionRequest { string weight_version = 1; }
+message UpdateWeightVersionResponse {}
+
+message GetWeightVersionRequest {}
+message GetWeightVersionResponse { string weight_version = 1; }
+
 // ======================================================================================
 // KV discovery
 // ======================================================================================

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

@@ -1,12 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
+use std::collections::BTreeMap;
 use std::sync::Arc;
 use std::time::Duration;
 
 use futures::future::{join_all, try_join_all};
 use itertools::Itertools;
 use serde::Serialize;
+use serde_json::Value as JsonValue;
 use tokio::sync::mpsc;
 use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, info, trace};
@@ -683,6 +685,77 @@ impl EngineCoreClient {
             .collect())
     }
 
+    /// Initialize the configured RL weight-transfer backend.
+    pub async fn init_weight_transfer_engine(&self, init_info: JsonValue) -> Result<()> {
+        self.collective_rpc(
+            "init_weight_transfer_engine",
+            None,
+            Vec::<JsonValue>::new(),
+            BTreeMap::from([("init_info".to_string(), init_info)]),
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// Start a weight update for the base model.
+    pub async fn start_weight_update(&self) -> Result<()> {
+        self.collective_rpc(
+            "start_weight_update",
+            None,
+            Vec::<JsonValue>::new(),
+            BTreeMap::<String, JsonValue>::new(),
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// Start a weight update for the speculative draft model.
+    pub async fn start_draft_weight_update(&self) -> Result<()> {
+        self.collective_rpc(
+            "start_draft_weight_update",
+            None,
+            Vec::<JsonValue>::new(),
+            BTreeMap::<String, JsonValue>::new(),
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// Apply one backend-specific weight metadata chunk.
+    pub async fn update_weights(&self, update_info: JsonValue) -> Result<()> {
+        self.collective_rpc(
+            "update_weights",
+            None,
+            Vec::<JsonValue>::new(),
+            BTreeMap::from([("update_info".to_string(), update_info)]),
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// Finish the current weight update.
+    pub async fn finish_weight_update(&self) -> Result<()> {
+        self.collective_rpc(
+            "finish_weight_update",
+            None,
+            Vec::<JsonValue>::new(),
+            BTreeMap::<String, JsonValue>::new(),
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// Set the committed weight version on every connected engine.
+    pub async fn set_weight_version(&self, weight_version: &str) -> Result<()> {
+        self.call_utility::<(), _>("set_weight_version", (weight_version,)).await?;
+        Ok(())
+    }
+
+    /// Return the committed weight version agreed on by every connected engine.
+    pub async fn get_weight_version(&self) -> Result<String> {
+        self.call_utility_consensus("get_weight_version", ()).await
+    }
+
     /// Return whether the engine is currently sleeping at any level.
     pub async fn is_sleeping(&self) -> Result<bool> {
         self.call_utility_consensus("is_sleeping", ()).await

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

@@ -67,6 +67,9 @@ pub fn default_ready_response() -> EngineCoreReadyResponse {
         kv_cache_size_tokens: None,
         kv_cache_max_concurrency: None,
         kv_events_config: None,
+        weight_transfer_backend: None,
+        enable_sleep_mode: false,
+        supports_draft_weight_updates: false,
     }
 }
 

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

@@ -87,6 +87,15 @@ pub struct EngineCoreReadyResponse {
     /// KV-event publisher configuration, if configured.
     #[serde(default)]
     pub kv_events_config: Option<KvEventsConfig>,
+    /// Configured RL weight-transfer backend, if weight transfer is enabled.
+    #[serde(default)]
+    pub weight_transfer_backend: Option<String>,
+    /// Whether the engine was started with sleep mode enabled.
+    #[serde(default)]
+    pub enable_sleep_mode: bool,
+    /// Whether the engine has a speculative draft model that can be updated.
+    #[serde(default)]
+    pub supports_draft_weight_updates: bool,
 }
 
 /// Frontend-owned ZMQ addresses that are sent to the engine during startup

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

@@ -2681,6 +2681,12 @@ fn python_msgpack_fixtures_match_rust_encoding() {
 
     let ready_response: EngineCoreReadyResponse =
         rmp_serde::from_slice(&hex::decode(ready_response_hex).unwrap()).unwrap();
+    assert_eq!(
+        ready_response.weight_transfer_backend.as_deref(),
+        Some("nccl")
+    );
+    assert!(ready_response.enable_sleep_mode);
+    assert!(ready_response.supports_draft_weight_updates);
     let kv_events_config = ready_response.kv_events_config.expect("KV events config should decode");
     assert!(kv_events_config.enable_kv_cache_events);
     assert_eq!(kv_events_config.publisher, "zmq");

rust/src/engine-core-client/src/tests/python_compat.py

@@ -387,6 +387,9 @@ class EngineCoreReadyResponse:
     kv_cache_size_tokens: int | None = None
     kv_cache_max_concurrency: float | None = None
     kv_events_config: KVEventsConfig | None = None
+    weight_transfer_backend: str | None = None
+    enable_sleep_mode: bool = False
+    supports_draft_weight_updates: bool = False
 
 
 ready_response = EngineCoreReadyResponse(
@@ -405,6 +408,9 @@ class EngineCoreReadyResponse:
     max_num_seqs=256,
     max_num_batched_tokens=8192,
     instance_id="test-instance",
+    weight_transfer_backend="nccl",
+    enable_sleep_mode=True,
+    supports_draft_weight_updates=True,
     kv_events_config=KVEventsConfig(
         enable_kv_cache_events=True,
         publisher="zmq",