feat(router): per-worker waiting-queue cap for least_load

bindings/python/src/lib.rs

@@ -513,6 +513,7 @@ struct Router {
     overlap_decay: f32,
     selection_temperature: f32,
     upstream_pool_idle_timeout_secs: u64,
+    least_load_max_waiting_requests: u32,
 }
 
 impl Router {
@@ -602,6 +603,7 @@ impl Router {
                     kv_pressure_weight: self.least_load_kv_pressure_weight,
                     mean_prefill_tokens: self.least_load_mean_prefill_tokens,
                     default_throughput: self.least_load_default_throughput,
+                    max_waiting_requests: self.least_load_max_waiting_requests,
                 },
                 PolicyType::Bucket => ConfigPolicyConfig::Bucket {
                     balance_abs_threshold: self.balance_abs_threshold,
@@ -1021,6 +1023,7 @@ impl Router {
         overlap_decay = 0.0,
         selection_temperature = 0.0,
         upstream_pool_idle_timeout_secs = 3,
+        least_load_max_waiting_requests = 0,
     ))]
     #[expect(clippy::too_many_arguments)]
     #[expect(
@@ -1158,6 +1161,7 @@ impl Router {
         overlap_decay: f32,
         selection_temperature: f32,
         upstream_pool_idle_timeout_secs: u64,
+        least_load_max_waiting_requests: u32,
     ) -> PyResult<Self> {
         let mut all_urls = worker_urls.clone();
 
@@ -1309,6 +1313,7 @@ impl Router {
             overlap_decay,
             selection_temperature,
             upstream_pool_idle_timeout_secs,
+            least_load_max_waiting_requests,
         })
     }
 

bindings/python/src/smg/router_args.py

@@ -213,6 +213,7 @@ class RouterArgs:
     overlap_decay: float = 0.0
     selection_temperature: float = 0.0
     upstream_pool_idle_timeout_secs: int = 3
+    least_load_max_waiting_requests: int = 0
 
     @staticmethod
     def add_cli_args(
@@ -463,6 +464,16 @@ def add_cli_args(
                 " request's token count is unknown at routing"
             ),
         )
+        routing_group.add_argument(
+            f"--{prefix}least-load-max-waiting-requests",
+            type=int,
+            default=RouterArgs.least_load_max_waiting_requests,
+            help=(
+                "Per-worker waiting-queue cap for least_load: skip workers whose"
+                " reported waiting requests (plus dispatches since their last"
+                " poll) have reached this count; 0 disables"
+            ),
+        )
         routing_group.add_argument(
             f"--{prefix}balance-abs-threshold",
             type=int,

bindings/python/tests/test_arg_parser.py

@@ -1222,6 +1222,7 @@ class TestRouterArgsFieldOrder:
         "overlap_decay",
         "selection_temperature",
         "upstream_pool_idle_timeout_secs",
+        "least_load_max_waiting_requests",
     ]
 
     def test_complete_field_sequence_is_frozen(self):
@@ -1243,6 +1244,7 @@ def test_new_fields_appended_after_positional_reserve(self):
             "overlap_decay",
             "selection_temperature",
             "upstream_pool_idle_timeout_secs",
+            "least_load_max_waiting_requests",
         ):
             assert names.index(appended) > marker, (
                 f"{appended} must be appended after worker_startup_delay to "

model_gateway/src/config/types.rs

@@ -555,6 +555,13 @@ pub enum PolicyConfig {
         /// per-replica generation rate; co-tunes with `kv_pressure_weight`.
         #[serde(default = "default_least_load_throughput")]
         default_throughput: f64,
+        /// Per-worker waiting-queue cap: skip workers whose reported waiting
+        /// requests, plus dispatches since their last poll, have reached this
+        /// count; when every candidate is at the cap, selection fails and the
+        /// request falls to the router's admission queue. `0` disables. Set
+        /// below the engine's max batch size.
+        #[serde(default)]
+        max_waiting_requests: u32,
     },
 
     #[serde(rename = "bucket")]

model_gateway/src/config/validation.rs

@@ -547,6 +547,7 @@ impl ConfigValidator {
                 kv_pressure_weight,
                 mean_prefill_tokens,
                 default_throughput,
+                max_waiting_requests: _,
             } => {
                 if *load_check_interval_secs == 0 {
                     return Err(ConfigError::InvalidValue {
@@ -1726,6 +1727,7 @@ mod tests {
                     kv_pressure_weight: 0.15,
                     mean_prefill_tokens: 1024,
                     default_throughput: 2000.0,
+                    max_waiting_requests: 0,
                 }),
                 prefill_policy: None,
                 decode_policy: None,

model_gateway/src/main.rs

@@ -297,6 +297,12 @@ struct CliArgs {
     #[arg(long, default_value_t = 1024, help_heading = "Routing Policy")]
     least_load_mean_prefill_tokens: u32,
 
+    /// Per-worker waiting-queue cap for least_load: skip workers whose reported
+    /// waiting requests (plus dispatches since their last poll) have reached
+    /// this count; 0 disables. Set below the engine's max batch size
+    #[arg(long, default_value_t = 0, help_heading = "Routing Policy")]
+    least_load_max_waiting_requests: u32,
+
     /// Enable data parallelism aware scheduling
     #[arg(long, default_value_t = false, help_heading = "Routing Policy")]
     dp_aware: bool,
@@ -1187,6 +1193,7 @@ impl CliArgs {
                 kv_pressure_weight: self.least_load_kv_pressure_weight,
                 mean_prefill_tokens: self.least_load_mean_prefill_tokens,
                 default_throughput: self.least_load_default_throughput,
+                max_waiting_requests: self.least_load_max_waiting_requests,
             },
             "bucket" => PolicyConfig::Bucket {
                 balance_abs_threshold: self.balance_abs_threshold,

model_gateway/src/policies/factory.rs

@@ -28,6 +28,7 @@ impl PolicyFactory {
                 kv_pressure_weight,
                 mean_prefill_tokens,
                 default_throughput,
+                max_waiting_requests,
                 ..
             } => {
                 // TODO: Pass load_check_interval_secs to WorkerMonitor for per-policy polling intervals.
@@ -36,6 +37,7 @@ impl PolicyFactory {
                     *kv_pressure_weight,
                     *mean_prefill_tokens,
                     *default_throughput,
+                    *max_waiting_requests,
                 ))
             }
             PolicyConfig::CacheAware {