feat(router): unify frontend and standalone selection core

.github/workflows/pre-merge.yml

@@ -435,6 +435,13 @@ jobs:
       working-directory: ${{ matrix.dir }}
       # NOTE: --all-targets doesn't run doc tests
       run: cargo test --locked --all-targets
+    - name: Check isolated kv-router feature builds
+      if: matrix.dir == '.'
+      working-directory: ${{ matrix.dir }}
+      # Workspace feature unification hides these; check each on its own.
+      run: |
+        cargo check --locked -p dynamo-kv-router --no-default-features --features standalone-slot-tracker
+        cargo check --locked -p dynamo-kv-router --no-default-features --features standalone-indexer
     - name: Run standalone slot-tracker feature tests
       if: matrix.dir == '.'
       working-directory: ${{ matrix.dir }}

deploy/inference-gateway/ext-proc/Cargo.toml

@@ -16,6 +16,7 @@ name = "dynamo-ext-proc"
 path = "src/bin/dynamo-ext-proc.rs"
 
 [dependencies]
+async-trait = { workspace = true }
 dynamo-kv-router = { workspace = true, features = [
     "metrics",
     "runtime-protocols",

deploy/inference-gateway/ext-proc/src/epp.rs

@@ -483,10 +483,6 @@ impl Router {
         allowed_worker_ids: Option<HashSet<u64>>,
         routing_constraints: RoutingConstraints,
     ) -> Result<PrefillReservation> {
-        if let Some(ref ids) = allowed_worker_ids {
-            self.prefill_router.register_workers(ids);
-        }
-
         self.prefill_router
             .reserve_prefill_worker(
                 reservation_id,
@@ -527,10 +523,6 @@ impl Router {
         allowed_worker_ids: Option<HashSet<u64>>,
         routing_constraints: RoutingConstraints,
     ) -> Result<(WorkerWithDpRank, u32)> {
-        if let Some(ref ids) = allowed_worker_ids {
-            self.decode_router.register_workers(ids);
-        }
-
         let config_override = decode_router_config_override(is_disaggregated);
 
         let outcome = self
@@ -934,8 +926,9 @@ fn indexed_endpoint_address(endpoint: &Endpoint) -> Option<String> {
 /// The mode is exclusive, and so are the identities. A worker process picks one
 /// `KubeDiscoveryTarget` from its own mode, so under container discovery
 /// nothing registers under the bare pod identity — emitting it there would
-/// invent a worker that `register_workers` upserts at zero load and zero KV
-/// overlap, making it the most attractive candidate the scheduler sees.
+/// put a worker id in `allowed_worker_ids` that no backend registered. The
+/// scheduler filters unknown ids out, so the pod contributes nothing, and a
+/// subset made only of such ids selects no worker at all.
 /// `"main"` hashes to the pod identity (`hash_container_name`), so a pod whose
 /// main container is Ready still contributes that id through the container
 /// path; one whose main container is *not* Ready correctly contributes nothing
@@ -1392,11 +1385,9 @@ impl EndpointPicker for Router {
             // Only pod discovery registers a worker under its pod identity.
             // Under container discovery each engine container registers under
             // its own (`KubeDiscoveryTarget::Container`), so a hand-built pod
-            // hash names a worker present in no registry: `register_workers`
-            // would upsert it at zero load and zero KV overlap, making it the
-            // scheduler's most attractive candidate, and the reverse lookup
-            // below would then fail to match and silently forward to
-            // `endpoints[0]`. The index is the one place that knows which
+            // hash names a worker present in no registry, which the scheduler
+            // filters out, leaving the subset short one candidate or empty.
+            // The index is the one place that knows which
             // identity scheme is in effect (see `pod_worker_ids`), so ask it.
             let wm: Vec<(u64, &Endpoint)> = {
                 let index = read_index(&self.worker_index);
@@ -2146,10 +2137,7 @@ mod tests {
     }
 
     /// Under pod discovery a worker registers under its pod identity alone, so
-    /// a pod's ready sidecars must contribute no worker ids. Emitting them
-    /// would invent workers no backend registered under: they miss
-    /// `register_workers`' discovery lookup, default to `(0, 1)` with no load
-    /// and no KV overlap, and so look maximally attractive to the scheduler.
+    /// a pod's ready sidecars must contribute no worker ids.
     #[test]
     fn pod_worker_ids_ignores_containers_under_pod_discovery() {
         let pod = pod_mode_worker_pod();
@@ -2215,7 +2203,7 @@ mod tests {
     }
 
     /// End of the chain that made this matter: the index feeds
-    /// `subset_to_worker_ids` -> `allowed_worker_ids` -> `register_workers`,
+    /// `subset_to_worker_ids` -> `allowed_worker_ids` -> the scheduler,
     /// so one backend pod must contribute exactly one worker id under pod
     /// discovery rather than one per ready container.
     #[test]
@@ -2450,8 +2438,7 @@ mod tests {
     /// An externally supplied endpoint must resolve to the identity the
     /// reflector actually holds. Under container discovery that is the engine
     /// container's id, never `hash_pod_name` -- deriving the latter names a
-    /// worker no registry contains, which `register_workers` then upserts at
-    /// zero load as the scheduler's most attractive candidate.
+    /// worker no registry contains.
     #[test]
     fn external_endpoint_resolves_to_the_indexed_container_identity() {
         let mut index = WorkerEndpointIndex::new(true);

deploy/inference-gateway/ext-proc/src/epp_router.rs

@@ -26,8 +26,9 @@ use std::time::Duration;
 use anyhow::Result;
 use tokio::sync::Semaphore;
 
-use dynamo_kv_router::services::selection::WorkerSelectionPolicyRegistry;
+use dynamo_kv_router::services::selection::{SelectionError, WorkerSelectionPolicyRegistry};
 use dynamo_llm::http::service::metadata::extract_metadata_from_header_pairs;
+use dynamo_llm::protocols::agents::HEADER_DYNAMO_SESSION_ID;
 use dynamo_llm::protocols::common::extensions::{
     AgentHints, HEADER_REQUEST_PRIORITY, HEADER_REQUEST_STRICT_PRIORITY, resolve_request_priority,
 };
@@ -368,6 +369,7 @@ impl EndpointPicker for EppRouter {
             model_name: self.model_name.clone(),
             reservation_id: reservation_id.clone(),
             token_ids: tokens,
+            session_id: first_header(&req.headers, HEADER_DYNAMO_SESSION_ID).map(str::to_owned),
             // `None` on the ordinary path: the selector schedules over its
             // catalog; `Some` only carries an Envoy subset constraint.
             allowed_worker_ids: allowed,
@@ -383,6 +385,9 @@ impl EndpointPicker for EppRouter {
 
         let resp = match self.selector.select_and_reserve(select_req).await {
             Ok(resp) => resp,
+            Err(SelectionError::BadRequest(message)) => {
+                return Err(PickError::InvalidRequest(message));
+            }
             Err(e) => return Err(PickError::RoutingFailed(e.to_string())),
         };
 

deploy/inference-gateway/ext-proc/src/epp_standalone_config.rs

@@ -140,6 +140,17 @@ pub struct EppStandaloneConfig {
     /// KV-cache block size; MUST equal the inference engine block size.
     #[validate(range(min = 1, message = "DYN_KV_CACHE_BLOCK_SIZE must be >= 1"))]
     pub block_size: u32,
+    /// Data-parallel ranks per worker pod. Only `1` is accepted: the EPP routes
+    /// by worker and cannot convey the selected rank to the pod.
+    #[validate(range(
+        min = 1,
+        max = 1,
+        message = "DYN_EPP_DATA_PARALLEL_SIZE must be 1; multi-rank standalone serving is unsupported"
+    ))]
+    pub data_parallel_size: u32,
+    /// Port distance between consecutive data-parallel ranks' KV event ports.
+    #[validate(range(min = 1, message = "DYN_EPP_KV_EVENT_PORT_STRIDE must be >= 1"))]
+    pub kv_event_port_stride: u16,
     /// KV zmq event port.
     #[validate(range(min = 1))]
     pub kv_event_port: u16,
@@ -163,6 +174,15 @@ pub struct EppStandaloneConfig {
     /// throughput throttle). Excess requests are shed with a 503, not queued.
     #[validate(range(min = 1, message = "DYN_EPP_MAX_INFLIGHT_REQUESTS must be >= 1"))]
     pub max_inflight_requests: usize,
+    /// Pin each `x-dynamo-session-id` to the worker that served it for this
+    /// long after its last request (`DYN_EPP_SESSION_AFFINITY_TTL_SECS`).
+    /// `None` disables session affinity.
+    #[validate(range(
+        min = 1.0,
+        max = 31536000.0,
+        message = "DYN_EPP_SESSION_AFFINITY_TTL_SECS must be between 1 and 31536000 seconds"
+    ))]
+    pub session_affinity_ttl_secs: Option<f64>,
 }
 
 impl EppStandaloneConfig {
@@ -216,13 +236,17 @@ impl EppStandaloneConfig {
             )?
             .unwrap_or(DEFAULT_TOKENIZER_MAX_RESPONSE_BYTES),
             block_size: opt_parse::<u32>(get, "DYN_KV_CACHE_BLOCK_SIZE")?.unwrap_or(0),
+            data_parallel_size: opt_parse::<u32>(get, "DYN_EPP_DATA_PARALLEL_SIZE")?.unwrap_or(1),
+            kv_event_port_stride: opt_parse::<u16>(get, "DYN_EPP_KV_EVENT_PORT_STRIDE")?
+                .unwrap_or(1),
             kv_event_port: opt_parse::<u16>(get, "DYN_EPP_KV_EVENT_PORT")?
                 .unwrap_or(DEFAULT_KV_EVENT_PORT),
             replay_port: opt_parse::<u16>(get, "DYN_EPP_KV_EVENT_REPLAY_PORT")?,
             total_kv_blocks: opt_parse::<u64>(get, "DYN_EPP_TOTAL_KV_BLOCKS")?,
             max_num_batched_tokens: opt_parse::<u64>(get, "DYN_EPP_MAX_NUM_BATCHED_TOKENS")?,
             max_inflight_requests: opt_parse::<usize>(get, "DYN_EPP_MAX_INFLIGHT_REQUESTS")?
                 .unwrap_or(DEFAULT_MAX_INFLIGHT_REQUESTS),
+            session_affinity_ttl_secs: opt_parse::<f64>(get, "DYN_EPP_SESSION_AFFINITY_TTL_SECS")?,
         })
     }
 
@@ -543,6 +567,21 @@ mod tests {
         );
     }
 
+    #[test]
+    fn data_parallel_size_above_one_fails() {
+        assert!(
+            parse_cfg(&[
+                ("DYN_EPP_INFERENCE_POOL_NAME", "vllm-qwen-pool"),
+                ("POD_NAMESPACE", "inference"),
+                ("DYN_MODEL_NAME", "Qwen/Qwen3-0.6B"),
+                ("DYN_EPP_TOKENIZER_SERVICE_URL", "http://vllm-render:8000"),
+                ("DYN_EPP_TOKENIZER_PROTOCOL", "vllm-render"),
+                ("DYN_EPP_DATA_PARALLEL_SIZE", "2"),
+            ])
+            .is_err()
+        );
+    }
+
     #[test]
     fn replay_port_is_optional_and_uses_the_kv_event_name() {
         let cfg = parse_cfg(&[

deploy/inference-gateway/ext-proc/src/lib.rs

@@ -39,7 +39,7 @@ pub use picker::{Endpoint, EndpointPicker, PickResult, RequestInfo, ResponseUsag
 pub use pod_discovery::{PodDiscovery, RawWorker};
 pub use render_http::RenderError;
 pub use runner::run;
-pub use selector::{OverlapSummary, SelectRequest, SelectResponse, Selector, WorkerRegistration};
+pub use selector::{OverlapSummary, SelectRequest, SelectResponse, Selector};
 pub use server::ExtProcServer;
 pub use sglang_renderer_client::SglangRendererClient;
 pub use topology_adapter::{RegistrationDefaults, TopologyAdapter};