feat: introduce RoutingConstraints + worker taints

lib/bindings/c/src/lib.rs

@@ -450,6 +450,7 @@ impl RouterHandles {
     ///
     /// When `allowed_worker_ids` is Some, only workers in that set are considered.
     /// Returns worker_id on success.
+    #[expect(clippy::too_many_arguments)]
     async fn query_prefill_worker(
         &self,
         tokens: &[u32],
@@ -458,6 +459,7 @@ impl RouterHandles {
         lora_name: Option<String>,
         priority_jump: f64,
         allowed_worker_ids: Option<HashSet<WorkerId>>,
+        routing_constraints: RoutingConstraints,
     ) -> Result<(u64, Option<u32>), QueryRouterResult> {
         if let Some(ref ids) = allowed_worker_ids {
             self.prefill_router.register_workers(ids);
@@ -471,6 +473,7 @@ impl RouterHandles {
                 lora_name,
                 priority_jump,
                 allowed_worker_ids,
+                routing_constraints,
             )
             .await
             .map_err(|e| {
@@ -500,6 +503,7 @@ impl RouterHandles {
         is_disaggregated: bool,
         priority_jump: f64,
         allowed_worker_ids: Option<HashSet<WorkerId>>,
+        routing_constraints: RoutingConstraints,
     ) -> Result<(WorkerWithDpRank, u32), QueryRouterResult> {
         if let Some(ref ids) = allowed_worker_ids {
             self.decode_router.register_workers(ids);
@@ -529,6 +533,7 @@ impl RouterHandles {
                 priority_jump,
                 None,
                 allowed_worker_ids,
+                routing_constraints,
             )
             .await
             .map_err(|e| {
@@ -1124,15 +1129,15 @@ pub unsafe extern "C" fn free_routing_result(result: *mut CRoutingResult) {
 /// Parse a JSON request string, apply the chat template, tokenize, and lift
 /// the router-relevant `priority_jump` out of `nvext.agent_hints.priority`.
 ///
-/// Returns `(token_ids, priority_jump)` on success, or a `QueryRouterResult`
+/// Returns `(token_ids, priority_jump, routing_constraints)` on success, or a `QueryRouterResult`
 /// error code. `priority_jump` is `0.0` when no hint is present. This mirrors
 /// the standalone Dynamo preprocessor lift in `lib/llm/src/preprocessor.rs`
 /// so the GAIE/EPP path produces the same queue ordering as a non-EPP
 /// deployment.
 unsafe fn preprocess_request(
     handles: &RouterHandles,
     request_json: *const c_char,
-) -> Result<(Vec<u32>, f64), QueryRouterResult> {
+) -> Result<(Vec<u32>, f64, RoutingConstraints), QueryRouterResult> {
     let preprocessor = match &handles.preprocessor {
         Some(p) => p,
         None => {
@@ -1156,6 +1161,11 @@ unsafe fn preprocess_request(
         };
 
     let priority_jump = extract_priority_jump(&request);
+    let routing_constraints = request
+        .nvext
+        .as_ref()
+        .and_then(|nvext| nvext.routing_constraints.clone())
+        .unwrap_or_default();
 
     let formatted_prompt = match preprocessor.apply_template(&request) {
         Ok(Some(prompt)) => prompt,
@@ -1182,7 +1192,7 @@ unsafe fn preprocess_request(
         "[EPP-TOKENIZE] Tokenized prompt in C bindings (this is the ONLY tokenization)"
     );
 
-    Ok((token_ids, priority_jump))
+    Ok((token_ids, priority_jump, routing_constraints))
 }
 
 /// Parse pods JSON into an optional set of allowed worker IDs.
@@ -1268,10 +1278,11 @@ pub unsafe extern "C" fn route_prefill_request(
 
     let handles = unsafe { &*handle };
 
-    let (tokens, priority_jump) = match unsafe { preprocess_request(handles, request_json) } {
-        Ok(t) => t,
-        Err(code) => return code,
-    };
+    let (tokens, priority_jump, routing_constraints) =
+        match unsafe { preprocess_request(handles, request_json) } {
+            Ok(t) => t,
+            Err(code) => return code,
+        };
 
     let allowed_worker_ids = unsafe { parse_pods_filter(pods_json) };
 
@@ -1284,6 +1295,7 @@ pub unsafe extern "C" fn route_prefill_request(
                 None,
                 priority_jump,
                 allowed_worker_ids,
+                routing_constraints,
             )
             .await?;
 
@@ -1346,16 +1358,23 @@ pub unsafe extern "C" fn route_decode_request(
 
     let handles = unsafe { &*handle };
 
-    let (tokens, priority_jump) = match unsafe { preprocess_request(handles, request_json) } {
-        Ok(t) => t,
-        Err(code) => return code,
-    };
+    let (tokens, priority_jump, routing_constraints) =
+        match unsafe { preprocess_request(handles, request_json) } {
+            Ok(t) => t,
+            Err(code) => return code,
+        };
 
     let allowed_worker_ids = unsafe { parse_pods_filter(pods_json) };
 
     let result = handles.runtime.secondary().block_on(async {
         let (decode_worker, _overlap_blocks) = handles
-            .query_decode_worker(&tokens, is_disaggregated, priority_jump, allowed_worker_ids)
+            .query_decode_worker(
+                &tokens,
+                is_disaggregated,
+                priority_jump,
+                allowed_worker_ids,
+                routing_constraints,
+            )
             .await?;
 
         tracing::info!(

lib/bindings/python/rust/lib.rs

@@ -39,7 +39,7 @@ use dynamo_llm::{self as llm_rs};
 
 use crate::llm::entrypoint::RouterConfig as PyRouterConfig;
 
-use crate::llm::local_model::ModelRuntimeConfig;
+use crate::llm::local_model::{ModelRuntimeConfig, RoutingConstraints};
 use crate::llm::preprocessor::{MediaDecoder, MediaFetcher};
 
 #[pyclass(eq, eq_int)]
@@ -183,6 +183,7 @@ fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<llm::kv::WorkerMetricsPublisher>()?;
     m.add_class::<llm::model_card::ModelDeploymentCard>()?; // Internal: only in _internal, not public API
     m.add_class::<llm::local_model::ModelRuntimeConfig>()?;
+    m.add_class::<RoutingConstraints>()?;
     m.add_class::<llm::preprocessor::MediaDecoder>()?;
     m.add_class::<llm::preprocessor::MediaFetcher>()?;
     m.add_class::<llm::kv::OverlapScores>()?;

lib/bindings/python/rust/llm/kv.rs

@@ -9,6 +9,7 @@ use std::sync::atomic::AtomicU32;
 use std::sync::mpsc;
 use tokio_stream::StreamExt;
 
+use super::local_model::RoutingConstraints;
 use super::*;
 use crate::Endpoint;
 #[cfg(feature = "kv-indexer")]
@@ -957,7 +958,7 @@ impl KvRouter {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[pyo3(signature = (token_ids, model, stop_conditions=None, sampling_options=None, output_options=None, router_config_override=None, worker_id=None, dp_rank=None, extra_args=None, block_mm_infos=None, multi_modal_data=None, mm_routing_info=None))]
+    #[pyo3(signature = (token_ids, model, stop_conditions=None, sampling_options=None, output_options=None, router_config_override=None, worker_id=None, dp_rank=None, extra_args=None, block_mm_infos=None, multi_modal_data=None, mm_routing_info=None, routing_constraints=None))]
     fn generate<'p>(
         &self,
         py: Python<'p>,
@@ -973,6 +974,7 @@ impl KvRouter {
         block_mm_infos: Option<PyObject>,
         multi_modal_data: Option<PyObject>,
         mm_routing_info: Option<PyObject>,
+        routing_constraints: Option<RoutingConstraints>,
     ) -> PyResult<Bound<'p, PyAny>> {
         // Depythonize the options with defaults
         let stop_conditions: StopConditions = if let Some(obj) = stop_conditions {
@@ -1048,10 +1050,11 @@ impl KvRouter {
             .tracker(Some(tracker.clone()));
 
         // Set routing hints if worker_id or dp_rank is provided
-        if worker_id.is_some() || dp_rank.is_some() {
+        if worker_id.is_some() || dp_rank.is_some() || routing_constraints.is_some() {
             let routing = llm_rs::protocols::common::preprocessor::RoutingHints {
                 backend_instance_id: worker_id,
                 dp_rank,
+                routing_constraints: routing_constraints.map(Into::into),
                 ..Default::default()
             };
             request_builder.routing(Some(routing));
@@ -1087,7 +1090,7 @@ impl KvRouter {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[pyo3(signature = (token_ids, router_config_override=None, request_id=None, update_indexer=false, block_mm_infos=None, lora_name=None))]
+    #[pyo3(signature = (token_ids, router_config_override=None, request_id=None, update_indexer=false, block_mm_infos=None, lora_name=None, routing_constraints=None))]
     fn best_worker<'p>(
         &self,
         py: Python<'p>,
@@ -1097,6 +1100,7 @@ impl KvRouter {
         update_indexer: bool,
         block_mm_infos: Option<PyObject>,
         lora_name: Option<String>,
+        routing_constraints: Option<RoutingConstraints>,
     ) -> PyResult<Bound<'p, PyAny>> {
         let router_config_override = if let Some(obj) = router_config_override {
             let override_config: RouterConfigOverride =
@@ -1125,6 +1129,7 @@ impl KvRouter {
                     0.0,
                     None,
                     None, // allowed_worker_ids: pass via RoutingHints in PreprocessedRequest path
+                    routing_constraints.map(Into::into).unwrap_or_default(),
                 )
                 .await
                 .map_err(to_pyerr)?;

lib/bindings/python/rust/llm/local_model.rs

@@ -1,10 +1,55 @@
 // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::collections::{HashMap, HashSet};
+
 use super::*;
+use dynamo_kv_router::protocols::RoutingConstraints as RsRoutingConstraints;
 use llm_rs::local_model::runtime_config::DisaggregatedEndpoint as RsDisaggregatedEndpoint;
 use llm_rs::local_model::runtime_config::ModelRuntimeConfig as RsModelRuntimeConfig;
 
+#[pyclass]
+#[derive(Clone, Debug, Default)]
+pub struct RoutingConstraints {
+    #[pyo3(get, set)]
+    pub required_taints: HashSet<String>,
+    #[pyo3(get, set)]
+    pub preferred_taints: HashMap<String, f32>,
+}
+
+#[pymethods]
+impl RoutingConstraints {
+    #[new]
+    #[pyo3(signature = (required_taints=None, preferred_taints=None))]
+    fn new(
+        required_taints: Option<HashSet<String>>,
+        preferred_taints: Option<HashMap<String, f32>>,
+    ) -> Self {
+        Self {
+            required_taints: required_taints.unwrap_or_default(),
+            preferred_taints: preferred_taints.unwrap_or_default(),
+        }
+    }
+}
+
+impl From<RoutingConstraints> for RsRoutingConstraints {
+    fn from(value: RoutingConstraints) -> Self {
+        Self {
+            required_taints: value.required_taints,
+            preferred_taints: value.preferred_taints,
+        }
+    }
+}
+
+impl From<RsRoutingConstraints> for RoutingConstraints {
+    fn from(value: RsRoutingConstraints) -> Self {
+        Self {
+            required_taints: value.required_taints,
+            preferred_taints: value.preferred_taints,
+        }
+    }
+}
+
 #[pyclass]
 #[derive(Clone, Debug, Default)]
 pub struct ModelRuntimeConfig {
@@ -73,6 +118,11 @@ impl ModelRuntimeConfig {
         self.inner.enable_eagle = enable_eagle;
     }
 
+    #[setter]
+    fn set_taints(&mut self, taints: HashSet<String>) {
+        self.inner.taints = taints;
+    }
+
     fn set_engine_specific(&mut self, key: &str, value: String) -> PyResult<()> {
         let value: serde_json::Value = serde_json::from_str(&value).map_err(to_pyerr)?;
         self.inner
@@ -182,4 +232,9 @@ impl ModelRuntimeConfig {
     fn enable_eagle(&self) -> bool {
         self.inner.enable_eagle
     }
+
+    #[getter]
+    fn taints(&self) -> HashSet<String> {
+        self.inner.taints.clone()
+    }
 }

lib/bindings/python/src/dynamo/_core.pyi

@@ -12,6 +12,7 @@ from typing import (
     List,
     Literal,
     Optional,
+    Set,
     Tuple,
 )
 
@@ -514,6 +515,7 @@ class ModelRuntimeConfig:
     data_parallel_size: int
     enable_local_indexer: bool
     enable_eagle: bool
+    taints: Set[str]
     runtime_data: dict[str, Any]
     tensor_model_config: Any | None
     bootstrap_host: str | None
@@ -545,6 +547,26 @@ class ModelRuntimeConfig:
         """Get the tensor model configuration."""
         ...
 
+class RoutingConstraints:
+    """
+    Request-side routing constraints.
+
+    ``required_taints`` is a hard eligibility filter.
+    ``preferred_taints`` maps taint -> signed weight.
+    Positive weights prefer matching workers, negative weights avoid them,
+    and ``0.0`` is neutral. Matching weights are summed and squashed with
+    ``tanh``, so opposite preferences cancel before Dynamo converts the
+    bounded bias into a strictly positive score multiplier.
+    """
+    required_taints: Set[str]
+    preferred_taints: Dict[str, float]
+
+    def __init__(
+        self,
+        required_taints: Optional[Set[str]] = None,
+        preferred_taints: Optional[Dict[str, float]] = None,
+    ) -> None: ...
+
 class OverlapScores:
     """
     A collection of prefix matching scores of workers for a given token ids.
@@ -1998,6 +2020,7 @@ class KvRouter:
         block_mm_infos: Optional[List[Optional[Dict[str, Any]]]] = None,
         multi_modal_data: Optional[JsonLike] = None,
         mm_routing_info: Optional[JsonLike] = None,
+        routing_constraints: Optional[RoutingConstraints] = None,
     ) -> AsyncIterator[JsonLike]:
         """
         Generate text using the KV-aware router.
@@ -2026,6 +2049,7 @@ class KvRouter:
             mm_routing_info: Optional structured routing-only multimodal payload
                             (e.g., {"routing_token_ids": [...], "block_mm_infos": [...]})
                             used by router selection without changing execution token_ids.
+            routing_constraints: Optional request routing constraints used to constrain or prefer tainted workers.
 
         Returns:
             An async iterator yielding generation responses
@@ -2059,6 +2083,7 @@ class KvRouter:
         update_indexer: bool = False,
         block_mm_infos: Optional[List[Optional[Dict[str, Any]]]] = None,
         lora_name: Optional[str] = None,
+        routing_constraints: Optional[RoutingConstraints] = None,
     ) -> Tuple[int, int, int]:
         """
         Find the best matching worker for the given tokens.

lib/bindings/python/src/dynamo/llm/__init__.py

@@ -34,6 +34,7 @@
 from dynamo._core import ReasoningConfig as ReasoningConfig
 from dynamo._core import RouterConfig as RouterConfig
 from dynamo._core import RouterMode as RouterMode
+from dynamo._core import RoutingConstraints as RoutingConstraints
 from dynamo._core import SglangArgs as SglangArgs
 from dynamo._core import WorkerMetricsPublisher as WorkerMetricsPublisher
 from dynamo._core import compute_block_hash_for_seq as compute_block_hash_for_seq