[Feature] Add EPD routing with JSON metadata

Cargo.toml

@@ -28,6 +28,7 @@ tower-http = { version = "0.6", features = [
 ] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
+sha2 = "0.10"
 bytes = "1.8.0"
 rand = "0.9.2"
 reqwest = { version = "0.13", features = ["stream", "blocking", "json"] }

docs/epd.md

@@ -0,0 +1,95 @@
+# Encoder + PD / E/P/D routing (experimental)
+
+The native Rust router can orchestrate remote encoders before forwarding a chat
+request to a combined prefill/decode worker or separate P and D workers. Embeddings move through vLLM's EC
+connector; the router only handles JSON placeholder metadata and transfer handles.
+No PyTorch dependency or tensor deserialization is needed in the router.
+
+Use vLLM with JSON metadata inputs (vllm-project/vllm#56090). Configure E workers
+as EC producers and PD workers as matching EC consumers, as for the vLLM encoder
+disaggregation example. The example connector needs a shared filesystem; NIXL
+and Mooncake need their usual backend dependencies and network connectivity.
+
+```bash
+cargo build --release --bin vllm-router
+target/release/vllm-router \
+  --host 0.0.0.0 --port 8000 \
+  --worker-urls http://pd0:8000 http://pd1:8000 \
+  --policy random \
+  --epd-config '{
+    "encoder_urls": ["http://e0:8000", "http://e1:8000"],
+    "consumer_zmq_addrs": {
+      "http://pd0:8000": "tcp://pd0:14579",
+      "http://pd1:8000": "tcp://pd1:14579"
+    }
+  }'
+```
+
+`consumer_zmq_addrs` is for Mooncake PUSH and must map each configured PD HTTP URL
+to that worker's configured EC control address. For NIXL PULL or the shared-file
+example connector, omit this map. The PD worker is selected before encoding so
+the producer and the HTTP request use the same destination. Encoder requests
+are distributed round-robin per media item.
+
+Send ordinary OpenAI chat requests to `/v1/chat/completions`. Each raw media item
+is sent to an encoder. Published metadata replaces that item with an embeds
+reference; missing metadata preserves the original media for vLLM's fallback.
+Repeated images keep their original positions and independent transfer IDs.
+Other request fields, including sampling options, pass through unchanged.
+Text-only and existing embeds inputs do not call the encoders. PD responses,
+including SSE streams, use the existing transparent forwarding path.
+
+Encoder failures are returned without submitting a partially prepared PD
+request. There is no automatic replay of EPD requests in this prototype.
+Abandoned Mooncake pushes receive best-effort cancellation over their existing
+ZMQ control channel; backend expiration remains the fallback if cancellation
+cannot reach the consumer. After a successful PD response header, the consumer
+owns request completion/abort cleanup.
+
+## Scope
+
+- Native binary CLI only; the Python launcher does not expose `--epd-config` yet.
+- Regular E+PD or static vLLM E/P/D mode, DP=1, no IGW.
+- Keep the existing backend's TP/PP constraints; this does not add parallelism support.
+- Static E URLs; E health-aware routing and dynamic consumer discovery are not added.
+- JSON metadata only, not legacy tensor-base64 metadata from old E servers.
+- No early metadata publication, encoder batching changes, or engine/scheduler changes.
+- Image E2E validation is performed separately; do not infer audio/video coverage
+  or production readiness from the metadata rewrite code alone.
+
+## Separate prefill and decode
+
+Add `--epd-config` to the existing static P/D router. Its EC control address map
+must reference **P**, not D. P is an EC consumer and KV producer; D is only a KV
+consumer. Both accept JSON placeholder metadata (`--enable-mm-embeds`).
+For Qwen3.5 with current vLLM NIXL KV transfer, P and D also require
+`VLLM_SSM_CONV_STATE_LAYOUT=DS` for Mamba state transfer.
+
+```bash
+target/release/vllm-router --host 0.0.0.0 --port 8000 \
+  --vllm-pd-disaggregation --kv-connector nixl \
+  --prefill http://p0:8000 --prefill http://p1:8000 \
+  --decode http://d0:8000 --decode http://d1:8000 --policy random \
+  --epd-config '{
+    "encoder_urls": ["http://e0:8000", "http://e1:8000"],
+    "consumer_zmq_addrs": {
+      "http://p0:8000": "tcp://p0:14579",
+      "http://p1:8000": "tcp://p1:14579"
+    }
+  }'
+```
+
+The router selects P/D, prepares E metadata for P, then uses the existing KV
+handoff. D receives the same placeholder layout with KV transfer parameters,
+without P's EC handles. P errors or missing NIXL KV handles stop the pipeline.
+Dynamic P/D discovery is not supported with EPD yet.
+
+```bash
+cargo test --test test_pd_routing test_epd_handoff
+```
+
+Focused tests:
+
+```bash
+cargo test --lib routers::http::epd::tests
+```

src/config/types.rs

@@ -6,6 +6,9 @@ use std::collections::HashMap;
 /// Main router configuration
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RouterConfig {
+    /// Optional encoder orchestration for E+PD or E/P/D routing.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub epd: Option<EpdConfig>,
     /// Routing mode configuration
     pub mode: RoutingMode,
     /// Worker connection mode
@@ -86,6 +89,16 @@ fn default_profile_timeout_secs() -> u64 {
     10
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct EpdConfig {
+    pub encoder_urls: Vec<String>,
+    /// Mooncake consumer control addresses, keyed by the P or combined PD HTTP URL.
+    /// Omit for pull-based NIXL or the shared-file example connector.
+    #[serde(default)]
+    pub consumer_zmq_addrs: HashMap<String, String>,
+}
+
 fn default_history_backend() -> HistoryBackend {
     HistoryBackend::Memory
 }
@@ -478,6 +491,7 @@ impl Default for RouterConfig {
             history_backend: default_history_backend(),
             enable_profiling: false,
             profile_timeout_secs: default_profile_timeout_secs(),
+            epd: None,
             kv_connector: KvConnector::default(),
         }
     }
@@ -1052,6 +1066,7 @@ mod tests {
             history_backend: default_history_backend(),
             enable_profiling: false,
             profile_timeout_secs: default_profile_timeout_secs(),
+            epd: None,
             kv_connector: KvConnector::default(),
         };
 
@@ -1118,6 +1133,7 @@ mod tests {
             history_backend: default_history_backend(),
             enable_profiling: false,
             profile_timeout_secs: default_profile_timeout_secs(),
+            epd: None,
             kv_connector: KvConnector::default(),
         };
 
@@ -1180,6 +1196,7 @@ mod tests {
             history_backend: default_history_backend(),
             enable_profiling: false,
             profile_timeout_secs: default_profile_timeout_secs(),
+            epd: None,
             kv_connector: KvConnector::default(),
         };
 

src/config/validation.rs

@@ -19,6 +19,7 @@ impl ConfigValidator {
         Self::validate_mode(&config.mode, has_service_discovery)?;
         Self::validate_policy(&config.policy)?;
         Self::validate_server_settings(config)?;
+        Self::validate_epd(config)?;
 
         if let Some(discovery) = &config.discovery {
             Self::validate_discovery(discovery, &config.mode)?;
@@ -39,6 +40,64 @@ impl ConfigValidator {
         Ok(())
     }
 
+    fn validate_epd(config: &RouterConfig) -> ConfigResult<()> {
+        let Some(epd) = &config.epd else {
+            return Ok(());
+        };
+        let worker_urls: Vec<&String> = match &config.mode {
+            RoutingMode::Regular { worker_urls } => worker_urls.iter().collect(),
+            RoutingMode::VllmPrefillDecode {
+                prefill_urls,
+                discovery_address: None,
+                ..
+            } => prefill_urls.iter().map(|(url, _)| url).collect(),
+            _ => {
+                return Err(ConfigError::ValidationFailed {
+                    reason: "EPD requires regular E+PD or static vLLM E/P/D routing".into(),
+                })
+            }
+        };
+        if config.intra_node_data_parallel_size != 1 || config.enable_igw {
+            return Err(ConfigError::ValidationFailed {
+                reason: "EPD currently requires DP=1 and proxy mode".into(),
+            });
+        }
+        if epd.encoder_urls.is_empty() {
+            return Err(ConfigError::MissingRequired {
+                field: "epd.encoder_urls".into(),
+            });
+        }
+        Self::validate_urls(&epd.encoder_urls)?;
+        if !epd.consumer_zmq_addrs.is_empty() {
+            if worker_urls.is_empty()
+                || epd.consumer_zmq_addrs.len() != worker_urls.len()
+                || worker_urls
+                    .iter()
+                    .any(|url| !epd.consumer_zmq_addrs.contains_key(*url))
+            {
+                return Err(ConfigError::ValidationFailed {
+                    reason:
+                        "EPD consumer_zmq_addrs must map every embedding consumer URL exactly once"
+                            .into(),
+                });
+            }
+            for addr in epd.consumer_zmq_addrs.values() {
+                let parsed = url::Url::parse(addr).map_err(|e| ConfigError::ValidationFailed {
+                    reason: format!("Invalid EC address: {e}"),
+                })?;
+                if parsed.scheme() != "tcp"
+                    || parsed.host_str().is_none()
+                    || parsed.port().is_none()
+                {
+                    return Err(ConfigError::ValidationFailed {
+                        reason: "EC addresses must be tcp://host:port".into(),
+                    });
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Validate routing mode configuration
     fn validate_mode(mode: &RoutingMode, has_service_discovery: bool) -> ConfigResult<()> {
         match mode {

src/lib.rs

@@ -176,6 +176,7 @@ impl Router {
         };
 
         Ok(config::RouterConfig {
+            epd: None,
             mode,
             policy,
             host: self.host.clone(),

src/main.rs

@@ -94,6 +94,9 @@ Examples:
 
 "#)]
 struct CliArgs {
+    /// E+PD or E/P/D encoder orchestration as JSON (encoder_urls, consumer_zmq_addrs).
+    #[arg(long)]
+    epd_config: Option<String>,
     /// Host address to bind the router server
     #[arg(long, default_value = "127.0.0.1")]
     host: String,
@@ -501,6 +504,14 @@ impl CliArgs {
 
         // Build RouterConfig
         Ok(RouterConfig {
+            epd: self
+                .epd_config
+                .as_deref()
+                .map(serde_json::from_str)
+                .transpose()
+                .map_err(|e| vllm_router_rs::config::ConfigError::ValidationFailed {
+                    reason: format!("Invalid --epd-config: {e}"),
+                })?,
             mode,
             policy,
             connection_mode,