[Router] Pin to the prefix owner when the whole fleet is queueing (--saturation-queue-floor) (#39169)

Co-authored-by: Kangyan Zhou <kangyan.zhou@radixark.ai>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

experimental/sgl-router/src/config/cli.rs

@@ -193,6 +193,19 @@ pub struct Cli {
     /// the published queue sums across a worker's DP ranks.
     #[arg(long)]
     pub worker_queue_limit: Option<u64>,
+    /// Saturation floor for `--worker-queue-limit` diversions: when no
+    /// cache candidate survives both the queue limit and hard admission,
+    /// at least one was over the limit, AND no worker in the routable
+    /// fleet has a fresh queue reading strictly below this floor, the
+    /// diverted request would wait wherever it lands, so it stays with the
+    /// least-pressured prefix owner instead — same wait, but prefilled
+    /// from cache instead of a full cold prefill that evicts other
+    /// prefixes and manufactures the next round of misses. Unset disables
+    /// the pin. Requires `--worker-queue-limit` (there is no diversion to
+    /// cancel without it) and must be at most the limit; scale with
+    /// `--dp-size` like the limit.
+    #[arg(long)]
+    pub saturation_queue_floor: Option<u64>,
 
     // ---- score composition ----
     /// Policies to sum, spelled exactly as `--policy` spells them and each
@@ -380,12 +393,35 @@ impl Cli {
             || self.cache_candidate_ratio.is_some()
             || self.cache_candidate_max_workers.is_some()
             || self.cache_switch_margin_tokens.is_some()
-            || self.worker_queue_limit.is_some();
+            || self.worker_queue_limit.is_some()
+            || self.saturation_queue_floor.is_some();
         // Value checks before the policy check: a value that is wrong under
         // every policy should say so, rather than pointing at --policy.
         if self.worker_queue_limit == Some(0) {
             return Err(anyhow!("--worker-queue-limit must be at least 1"));
         }
+        if let Some(floor) = self.saturation_queue_floor {
+            // The floor modifies the gate's diversion; without the gate
+            // there is no diversion to cancel and the knob would sit dead.
+            let Some(limit) = self.worker_queue_limit else {
+                return Err(anyhow!(
+                    "--saturation-queue-floor requires --worker-queue-limit (there is no \
+                     diversion to cancel without it)"
+                ));
+            };
+            if floor == 0 {
+                return Err(anyhow!("--saturation-queue-floor must be at least 1"));
+            }
+            // floor <= limit keeps the saturation label readable: a floor
+            // above the limit would declare the fleet saturated while
+            // workers the gate still admits exist.
+            if floor > limit {
+                return Err(anyhow!(
+                    "--saturation-queue-floor ({floor}) must be at most --worker-queue-limit \
+                     ({limit})"
+                ));
+            }
+        }
         if tuned_cache_candidates && self.policy != PolicyKind::CacheAware {
             return Err(anyhow!(
                 "cache candidate tuning flags require --policy cache_aware"
@@ -610,6 +646,7 @@ impl Cli {
                     .cache_switch_margin_tokens
                     .unwrap_or(d.cache_switch_margin_tokens),
                 worker_queue_limit: self.worker_queue_limit.or(d.worker_queue_limit),
+                saturation_queue_floor: self.saturation_queue_floor.or(d.saturation_queue_floor),
             })
         } else {
             None
@@ -1972,6 +2009,74 @@ mod tests {
         assert!(err.contains("--worker-queue-limit"), "got: {err}");
     }
 
+    #[test]
+    fn saturation_queue_floor_requires_the_queue_gate_and_stays_below_it() {
+        let config = cfg_of(
+            "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \
+             --worker-queue-limit 4 --saturation-queue-floor 2",
+        )
+        .unwrap();
+        assert_eq!(
+            config
+                .model
+                .affinity
+                .expect("cache-aware needs affinity config")
+                .saturation_queue_floor,
+            Some(2)
+        );
+
+        // Unset, the pin is disabled and the gate behaves as before.
+        let defaults = cfg_of(
+            "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \
+             --worker-queue-limit 4",
+        )
+        .unwrap();
+        assert_eq!(
+            defaults
+                .model
+                .affinity
+                .expect("default affinity config")
+                .saturation_queue_floor,
+            None
+        );
+
+        // Without the gate there is no diversion to cancel.
+        let err = cfg_of(
+            "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \
+             --saturation-queue-floor 2",
+        )
+        .expect_err("the floor modifies the gate's diversion")
+        .to_string();
+        assert!(err.contains("--saturation-queue-floor"), "got: {err}");
+        assert!(err.contains("--worker-queue-limit"), "got: {err}");
+
+        // A floor above the limit would declare saturation while workers
+        // the gate still admits exist.
+        let err = cfg_of(
+            "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \
+             --worker-queue-limit 4 --saturation-queue-floor 5",
+        )
+        .expect_err("floor must not exceed the limit")
+        .to_string();
+        assert!(err.contains("at most"), "got: {err}");
+
+        let err = cfg_of(
+            "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \
+             --worker-queue-limit 4 --saturation-queue-floor 0",
+        )
+        .expect_err("a zero floor would reject every queue reading")
+        .to_string();
+        assert!(err.contains("--saturation-queue-floor"), "got: {err}");
+
+        let err = cfg_of("--policy power_of_two --worker-queue-limit 4 --saturation-queue-floor 2")
+            .expect_err("the pin only governs cache-affinity selection")
+            .to_string();
+        assert!(
+            err.contains("cache candidate tuning flags require --policy cache_aware"),
+            "got: {err}"
+        );
+    }
+
     #[test]
     fn cache_candidate_cli_rejects_invalid_bounds() {
         for (args, expected) in [

experimental/sgl-router/src/config/types.rs

@@ -534,7 +534,31 @@ pub struct AffinityConfig {
     /// Note the firing point scales with `dp_size`: the sample sums `waiting`
     /// across a worker's DP ranks while a request lands on one of them, so
     /// scale the limit with `--dp-size` on DP-attention deployments.
+    ///
+    /// The companion `saturation_queue_floor` cancels the gate's diversions
+    /// when they have no payoff (nothing in the fleet reads below the
+    /// floor).
     pub worker_queue_limit: Option<u64>,
+    /// Saturation pin (`--saturation-queue-floor`): cancels queue-gate
+    /// diversions that have no payoff. When no cache candidate survives
+    /// both `worker_queue_limit` and hard admission, at least one was over
+    /// the limit, AND no worker in the routable fleet has a fresh queue
+    /// reading strictly below this floor, the diverted request would wait
+    /// wherever it lands — so it pins to the least-pressured prefix owner
+    /// instead of cold-prefilling on a non-owner (which evicts other
+    /// prefixes and manufactures the next round of misses). `None` — the
+    /// default — preserves the pure gate behavior.
+    ///
+    /// Polarity note: a worker with no fresh sample does NOT count as idle
+    /// — the opposite of the gate's fail-open, and deliberately so. The
+    /// gate keeps affinity because that is the safe default action; the
+    /// pin asks whether a *provably better* destination exists, and an
+    /// unknown queue is not proof. Both polarities leave the request with
+    /// its prefix owner when the signal is missing.
+    ///
+    /// The CLI enforces `floor <= worker_queue_limit` and requires the
+    /// gate; like the limit, scale the floor with `dp_size`.
+    pub saturation_queue_floor: Option<u64>,
 }
 
 impl Default for AffinityConfig {
@@ -558,6 +582,7 @@ impl Default for AffinityConfig {
             cache_candidate_max_workers: 32,
             cache_switch_margin_tokens: 1_024,
             worker_queue_limit: None,
+            saturation_queue_floor: None,
         }
     }
 }

experimental/sgl-router/src/policies/admission.rs

@@ -11,6 +11,11 @@
 //! that reads [`EngineWorkerLoad::num_waiting_reqs`] rather than the native
 //! monitor fields: a worker already making requests wait cannot win on cache
 //! affinity. It fails open on a missing sample — see [`queue_gate_admits`].
+//! The optional saturation floor (`--saturation-queue-floor`) cancels a
+//! diversion that has no payoff: when no candidate survives the gate and hard
+//! admission, at least one was gate-rejected, and no worker in the routable
+//! fleet reads below the floor, the request pins to the least-pressured
+//! prefix owner instead of cold-prefilling on a non-owner.
 
 use crate::policies::engine_load::{EngineLoadSnapshot, EngineWorkerLoad, NativeCacheWorkerLoad};
 use crate::policies::power_of_two::select_with_snapshot;
@@ -113,6 +118,11 @@ pub enum DecisionReason {
     BackupPressureGuard,
     RangeFallback,
     CapacityFallbackPowerOfTwo,
+    /// Fleet saturated: no cache candidate survived the queue gate and
+    /// capacity admission, at least one was rejected by the gate, and no
+    /// fleet worker has a fresh queue reading below the saturation floor —
+    /// so the request pinned to a prefix owner instead of diverting.
+    SaturationPin,
 }
 
 #[derive(Clone)]
@@ -198,6 +208,11 @@ pub(crate) fn fleet_is_all_queued(
 }
 
 /// Selects a worker from bounded cache candidates and records guard coverage.
+/// `fleet` is the worker set this request could actually be routed to (the
+/// model's healthy prefill pool). Only the saturation pin reads it, and it
+/// must be the routable fleet rather than the router-wide load table: that
+/// table also holds decode peers and other models' workers, none of which a
+/// diversion could reach.
 pub fn resolve_cache_candidates(
     proposal: &CacheCandidateProposal,
     request_input_tokens: u64,
@@ -212,24 +227,36 @@ pub fn resolve_cache_candidates(
     // the gate answer per candidate is what splits the set, so asking twice
     // would re-read the snapshot for every candidate on every request.
     let mut evaluated: Vec<&CacheCandidate> = Vec::with_capacity(proposal.candidates.len());
-    let mut queue_gate_rejected_candidates = 0u64;
+    // Kept, not just counted: the saturation pin ranks these by pressure when
+    // it fires. Stays unallocated while the gate is disabled, because nothing
+    // is ever rejected then.
+    let mut queue_gate_rejected: Vec<&CacheCandidate> = Vec::new();
     let mut queue_gate_best_rejected_blocks = 0u32;
     for candidate in &proposal.candidates {
         if queue_gate_admits(snapshot, &candidate.worker, queue_limit) {
             evaluated.push(candidate);
         } else {
-            queue_gate_rejected_candidates += 1;
+            queue_gate_rejected.push(candidate);
             queue_gate_best_rejected_blocks =
                 queue_gate_best_rejected_blocks.max(candidate.matched_prefix_blocks);
         }
     }
+    let queue_gate_rejected_candidates = queue_gate_rejected.len() as u64;
     // Second tier, mirroring `range_fallback`: when the gate removed every
     // candidate AND nowhere in the fleet is unqueued, diversion cannot dodge
     // a wait, so returning no decision would trade the whole prefix for
     // nothing. Re-admit the ungated set. While an unqueued worker still
     // exists the gate keeps its teeth and the request leaves the prefix.
-    let queue_gate_fell_back =
-        evaluated.is_empty() && queue_gate_rejected_candidates > 0 && fleet_all_queued;
+    //
+    // A configured saturation floor supersedes this tier rather than stacking
+    // with it: the floor names a weaker, tunable saturation condition and
+    // pins with a pressure-only ranking, where re-admission would re-rank by
+    // uncached work first. Both keep the prefix; only one may decide which
+    // owner, so the explicit knob wins and this tier covers the unset case.
+    let queue_gate_fell_back = evaluated.is_empty()
+        && queue_gate_rejected_candidates > 0
+        && fleet_all_queued
+        && proposal.saturation_queue_floor.is_none();
     if queue_gate_fell_back {
         evaluated.extend(proposal.candidates.iter());
     }
@@ -253,8 +280,60 @@ pub fn resolve_cache_candidates(
         .copied()
         .min_by_key(|candidate| candidate.uncached_tokens)
     else {
+        // Saturation pin: no candidate survived the gate and hard
+        // admission (and at least one was gate-rejected), but diverting
+        // only pays when a meaningfully idle destination exists. With a
+        // floor configured and no fresh queue reading below it, the request
+        // would wait wherever it lands — so waiting at a prefix owner
+        // dominates: same wait, prefill from cache instead of a full cold
+        // prefill that evicts other prefixes and manufactures the next
+        // round of misses. Saturation suspends the gate, not the tiebreak:
+        // pin to the least-pressured rejected owner, skipping any that also
+        // fail hard admission (a capacity-exhausted owner cannot take the
+        // request).
+        let pinned = if queue_gate_rejected.is_empty() {
+            None
+        } else {
+            proposal.saturation_queue_floor.and_then(|floor| {
+                if snapshot
+                    .any_fresh_queue_below(fleet.iter().map(|worker| worker.url.as_str()), floor)
+                {
+                    return None;
+                }
+                // Ranked over its own lookup, not `loads`: `loads` covers the
+                // gate-ADMITTED set, which is empty precisely when the pin
+                // fires. An empty lookup reports no engine coverage, so every
+                // comparison would fall back to router-local load — tie at
+                // zero for every owner, decided by worker id. The pin ranks
+                // the rejected owners, so it must see the rejected owners.
+                let pin_loads = FreshLoadLookup::new(
+                    Some(snapshot),
+                    queue_gate_rejected
+                        .iter()
+                        .map(|candidate| &candidate.worker),
+                );
+                queue_gate_rejected
+                    .iter()
+                    .copied()
+                    .filter(|candidate| {
+                        is_cache_candidate_admitted(candidate, request_input_tokens, &pin_loads)
+                    })
+                    .min_by(|left, right| {
+                        pin_loads
+                            .compare_prefill_pressure(&left.worker, &right.worker)
+                            .then_with(|| left.worker.id.0.cmp(&right.worker.id.0))
+                    })
+            })
+        };
         return CacheCandidateResolution {
-            decision: None,
+            decision: pinned.map(|pinned| FinalDecision {
+                selected: Arc::clone(&pinned.worker),
+                primary: Arc::clone(&pinned.worker),
+                backup: None,
+                reason: DecisionReason::SaturationPin,
+                candidate_range_id: pinned.candidate_range_id.clone(),
+                load_snapshot_version: snapshot.version,
+            }),
             prefill_pressure_source: loads.prefill_pressure_source(),
             admission_evaluated_candidates,
             admission_rejected_candidates,
@@ -1132,6 +1211,7 @@ mod tests {
             pressure_abs_threshold_ms: None,
             pressure_rel_threshold: 1.5,
             worker_queue_limit: None,
+            saturation_queue_floor: None,
         };
         let loads = snapshot(&[
             (&congested, 1, 1_000, 10, 10_000),
@@ -1179,6 +1259,19 @@ mod tests {
         }
     }
 
+    fn saturation_proposal(
+        candidates: Vec<CacheCandidate>,
+        limit: Option<u64>,
+        floor: Option<u64>,
+    ) -> CacheCandidateProposal {
+        CacheCandidateProposal {
+            candidates,
+            worker_queue_limit: limit,
+            saturation_queue_floor: floor,
+            ..Default::default()
+        }
+    }
+
     #[test]
     fn queue_gate_diverts_off_an_owner_over_the_limit() {
         let owner = worker("owner");
@@ -1398,6 +1491,178 @@ mod tests {
         assert!(!resolution.queue_gate_fell_back);
     }
 
+    #[test]
+    fn saturation_pin_keeps_affinity_with_the_least_pressured_owner() {
+        let calm_owner = worker("calm_owner");
+        let busy_owner = worker("busy_owner");
+        // A fleet worker that is NOT a cache candidate: the saturation check
+        // reads fleet-wide fresh samples, not just the candidate set. It is
+        // over the floor, so it does not break the saturation claim.
+        let fleet_only = worker("fleet_only");
+        let proposal = saturation_proposal(
+            vec![
+                // The busy owner holds the deeper prefix and would win
+                // without the gate; both owners are over the limit.
+                candidate(&busy_owner, 10, 9),
+                candidate(&calm_owner, 60, 4),
+            ],
+            Some(4),
+            Some(2),
+        );
+        let loads = snapshot(&[
+            (&busy_owner, 1, 9, 10, 10_000),
+            (&calm_owner, 1, 5, 10, 10_000),
+            (&fleet_only, 1, 3, 10, 10_000),
+        ]);
+        let fleet = vec![
+            Arc::clone(&busy_owner),
+            Arc::clone(&calm_owner),
+            Arc::clone(&fleet_only),
+        ];
+
+        let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet);
+
+        let decision = resolution
+            .decision
+            .expect("nothing reads below the floor: pin to a prefix owner");
+        assert_eq!(decision.reason, DecisionReason::SaturationPin);
+        assert_eq!(
+            decision.selected.id, calm_owner.id,
+            "saturation suspends the gate, not the tiebreak"
+        );
+        assert_eq!(decision.primary.id, calm_owner.id);
+        assert!(decision.backup.is_none());
+        assert_eq!(decision.load_snapshot_version, loads.version);
+        assert_eq!(resolution.queue_gate_rejected_candidates, 2);
+        assert_eq!(resolution.queue_gate_best_rejected_blocks, 9);
+        assert_eq!(resolution.admission_rejected_candidates, 0);
+    }
+
+    #[test]
+    fn no_saturation_floor_preserves_queue_gate_exhaustion() {
+        let owner = worker("owner");
+        let other = worker("other");
+        let proposal = saturation_proposal(
+            vec![candidate(&owner, 10, 9), candidate(&other, 60, 4)],
+            Some(4),
+            None,
+        );
+        let loads = snapshot(&[(&owner, 1, 9, 10, 10_000), (&other, 1, 5, 10, 10_000)]);
+        let fleet = vec![Arc::clone(&owner), Arc::clone(&other)];
+
+        let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet);
+
+        // With no floor the queue gate's own second tier applies: every owner
+        // is over the limit and nowhere in the fleet is unqueued, so the
+        // prefix is kept rather than traded for a wait that cannot be dodged.
+        // The pin is what a floor buys; without one this is the behaviour.
+        assert!(resolution.queue_gate_fell_back);
+        assert_eq!(
+            resolution.decision.as_ref().map(|d| d.reason),
+            Some(DecisionReason::CacheCandidate)
+        );
+        assert_eq!(resolution.queue_gate_rejected_candidates, 2);
+        assert_eq!(resolution.queue_gate_best_rejected_blocks, 9);
+        assert_eq!(resolution.admission_rejected_candidates, 0);
+    }
+
+    #[test]
+    fn saturation_pin_yields_to_a_provably_idle_fleet_worker() {
+        let owner = worker("owner");
+        // Not a candidate — but a fresh reading below the floor anywhere in
+        // the fleet means diverting can pay, so the pin must not fire.
+        let idle_elsewhere = worker("idle_elsewhere");
+        let proposal = saturation_proposal(vec![candidate(&owner, 10, 9)], Some(4), Some(2));
+        let loads = snapshot(&[
+            (&owner, 1, 9, 10, 10_000),
+            (&idle_elsewhere, 1, 0, 10, 10_000),
+        ]);
+        let fleet = vec![Arc::clone(&owner), Arc::clone(&idle_elsewhere)];
+
+        let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet);
+
+        assert!(resolution.decision.is_none());
+        assert_eq!(resolution.queue_gate_rejected_candidates, 1);
+        assert_eq!(resolution.queue_gate_best_rejected_blocks, 9);
+    }
+
+    #[test]
+    fn saturation_pin_treats_an_unknown_queue_as_not_idle() {
+        let owner = worker("owner");
+        // In the fleet, routable, and never published a sample.
+        let unsampled = worker("unsampled");
+        let proposal = saturation_proposal(vec![candidate(&owner, 10, 9)], Some(4), Some(2));
+        // The snapshot holds only the over-limit owner. `unsampled` is a
+        // real destination whose queue is unknown, not proof of a better
+        // one — opposite of the gate's fail-open.
+        let loads = snapshot(&[(&owner, 1, 9, 10, 10_000)]);
+        let fleet = vec![Arc::clone(&owner), Arc::clone(&unsampled)];
+
+        let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet);
+
+        let decision = resolution
+            .decision
+            .expect("an unknown queue must not read as below the floor");
+        assert_eq!(decision.reason, DecisionReason::SaturationPin);
+        assert_eq!(decision.selected.id, owner.id);
+    }
+
+    #[test]
+    fn saturation_pin_ignores_idle_workers_outside_the_routable_fleet() {
+        let owner = worker("owner");
+        // Present in the router-wide load table but not routable for this
+        // request: a PD decode peer, another model's worker, or a worker the
+        // registry no longer reports healthy. Decode peers idle near zero
+        // waiting, so scanning the whole table would veto the pin on every
+        // PD deployment.
+        let off_fleet_idle = worker("off_fleet_idle");
+        let proposal = saturation_proposal(vec![candidate(&owner, 10, 9)], Some(4), Some(2));
+        let loads = snapshot(&[
+            (&owner, 1, 9, 10, 10_000),
+            (&off_fleet_idle, 1, 0, 10, 10_000),
+        ]);
+        let fleet = vec![Arc::clone(&owner)];
+
+        let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet);
+
+        let decision = resolution
+            .decision
+            .expect("an unroutable idle worker is not a destination a diversion could reach");
+        assert_eq!(decision.reason, DecisionReason::SaturationPin);
+        assert_eq!(decision.selected.id, owner.id);
+    }
+
+    #[test]
+    fn saturation_pin_skips_a_capacity_exhausted_owner() {
+        let full = worker("full");
+        let admitted_owner = worker("admitted_owner");
+        let proposal = saturation_proposal(
+            vec![candidate(&full, 10, 9), candidate(&admitted_owner, 60, 4)],
+            Some(4),
+            Some(2),
+        );
+        // Both owners are over the queue limit, and `full` is the pressure
+        // minimum — but it is also KV-exhausted (used + request exceeds
+        // capacity), so it cannot take the request even pinned.
+        let loads = snapshot(&[
+            (&full, 1, 5, 10_000, 10_000),
+            (&admitted_owner, 1, 9, 10, 10_000),
+        ]);
+        let fleet = vec![Arc::clone(&full), Arc::clone(&admitted_owner)];
+
+        let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet);
+
+        let decision = resolution
+            .decision
+            .expect("the capacity-admitted owner can be pinned");
+        assert_eq!(decision.reason, DecisionReason::SaturationPin);
+        assert_eq!(decision.selected.id, admitted_owner.id);
+        // `full` is booked under the gate, not capacity — the pin's own
+        // capacity filter must not pollute the audit counters.
+        assert_eq!(resolution.queue_gate_rejected_candidates, 2);
+        assert_eq!(resolution.admission_rejected_candidates, 0);
+    }
+
     #[test]
     fn range_fallback_prefers_an_unqueued_worker_over_a_shallower_queueing_one() {
         // The primary fails KV-capacity admission so selection reaches the

experimental/sgl-router/src/policies/cache_aware.rs

@@ -110,6 +110,7 @@ impl CacheAwarePolicy {
             pressure_abs_threshold_ms: self.config.pressure_abs_threshold_ms,
             pressure_rel_threshold: self.config.pressure_rel_threshold,
             worker_queue_limit: self.config.worker_queue_limit,
+            saturation_queue_floor: self.config.saturation_queue_floor,
         })
     }
 

experimental/sgl-router/src/policies/engine_load.rs

@@ -105,6 +105,33 @@ impl EngineLoadSnapshot {
         self.workers.get(worker_url)
     }
 
+    /// True when some worker in `fleet_urls` has a fresh queue reading
+    /// strictly below `floor` — a destination where a diverted request would
+    /// provably wait behind fewer than `floor` others. Workers with no fresh
+    /// sample do not count: the saturation pin asks whether a provably better
+    /// destination exists, and an unknown queue is not proof. This is the
+    /// opposite polarity from the queue gate's fail-open, and the two
+    /// compose: both leave the request with its prefix owner when the signal
+    /// is missing.
+    ///
+    /// The caller passes the routable fleet rather than letting this scan the
+    /// whole table: the table is router-wide, so it also holds decode peers,
+    /// other models' workers, and workers no longer healthy enough to receive
+    /// this request. None of those is a destination a diversion could reach,
+    /// and a decode peer idling at zero waiting would otherwise veto the pin
+    /// on every PD deployment.
+    pub fn any_fresh_queue_below<'u>(
+        &self,
+        fleet_urls: impl IntoIterator<Item = &'u str>,
+        floor: u64,
+    ) -> bool {
+        fleet_urls.into_iter().any(|url| {
+            self.workers
+                .get(url)
+                .is_some_and(|load| load.num_waiting_reqs < floor)
+        })
+    }
+
     /// Returns only complete, fresh native Cache-Aware monitor data.
     pub fn fresh_native_cache_load_for_url(
         &self,

experimental/sgl-router/src/policies/mod.rs

@@ -356,6 +356,14 @@ pub struct CacheCandidateProposal {
     /// waiting requests cannot win on cache affinity. `None` disables the
     /// gate. See [`crate::config::AffinityConfig::worker_queue_limit`].
     pub worker_queue_limit: Option<u64>,
+    /// Saturation pin: when no candidate survives the gate and hard
+    /// admission, at least one was queue-gate-rejected, and no worker in
+    /// the routable fleet has a fresh queue reading strictly below this
+    /// floor, the request pins to the least-pressured rejected prefix
+    /// owner instead of diverting — the diversion cannot dodge a wait and
+    /// would forfeit the matched prefix. `None` disables the pin. See
+    /// [`crate::config::AffinityConfig::saturation_queue_floor`].
+    pub saturation_queue_floor: Option<u64>,
 }
 
 /// Prefill proposal returned as either a pair or a Cache-Aware candidate set.

experimental/sgl-router/src/policies/selection.rs

@@ -70,6 +70,8 @@ pub(crate) struct PrefillSelectionInputs<'a> {
     pub session_affinity_mode: SessionAffinityMode,
     /// `--worker-queue-limit`. `None` disables the queue gate entirely.
     pub worker_queue_limit: Option<u64>,
+    /// `--saturation-queue-floor`. `None` disables the saturation pin.
+    pub saturation_queue_floor: Option<u64>,
 }
 
 /// The queue-gate blind warn is sampled: it fires on a per-request path, and
@@ -78,6 +80,13 @@ pub(crate) struct PrefillSelectionInputs<'a> {
 const QUEUE_GATE_BLIND_LOG_SAMPLE: u64 = 64;
 static QUEUE_GATE_BLIND_LOG_COUNTER: AtomicU64 = AtomicU64::new(0);
 
+/// The saturation-pin info log is sampled for the same reason as the
+/// queue-gate blind warn: it fires on a per-request path and the condition
+/// (a saturated fleet) persists for many requests, so 1-in-64 surfaces it
+/// without log flooding.
+const SATURATION_PIN_LOG_SAMPLE: u64 = 64;
+static SATURATION_PIN_LOG_COUNTER: AtomicU64 = AtomicU64::new(0);
+
 /// Maps the queue-gate audit of a Cache-Aware selection that produced no
 /// winner onto its decision label. Pure so every boundary is pinned by unit
 /// tests rather than inferred from the ladder that calls it:
@@ -354,21 +363,52 @@ impl<'a> Selector<'a> {
             prefill_pressure_source = cache_decision.prefill_pressure_source,
             "cache candidate winner",
         );
-        inputs
-            .metrics
-            .record_policy_decision("cache_aware", "cache_candidate");
-        inputs.metrics.record_cache_aware_decision(
-            &inputs.model_id.0,
-            if cache_decision.queue_gate_fell_back {
-                // The gate removed every owner and nowhere in the fleet is
-                // unqueued, so the prefix was kept rather than traded for a
-                // wait that cannot be dodged. Booked as saturation, never as
-                // a plain hit.
-                CacheAwareDecision::AllQueued
-            } else {
-                CacheAwareDecision::CacheHit
-            },
+        inputs.metrics.record_policy_decision(
+            "cache_aware",
+            prefill_policy_reason(
+                PolicyKind::CacheAware,
+                ProposalKind::CacheAffinity,
+                decision.reason,
+                inputs.session_id.is_some_and(|value| !value.is_empty()),
+                true,
+            ),
         );
+        if decision.reason == DecisionReason::SaturationPin {
+            // The pin books the saturation label because it always means
+            // affinity was kept under a queueing fleet. It does not retire
+            // the off-owner draw in `run`: when every gate-rejected owner
+            // also fails capacity admission the pin yields no decision, and
+            // the fallback records the same label from an off-owner landing.
+            inputs
+                .metrics
+                .record_cache_aware_decision(&inputs.model_id.0, CacheAwareDecision::AllQueued);
+            if SATURATION_PIN_LOG_COUNTER
+                .fetch_add(1, AtomicOrdering::Relaxed)
+                .is_multiple_of(SATURATION_PIN_LOG_SAMPLE)
+            {
+                tracing::info!(
+                    model = %&inputs.model_id.0,
+                    worker = %decision.selected.url,
+                    saturation_queue_floor = inputs.saturation_queue_floor,
+                    worker_queue_limit = inputs.worker_queue_limit,
+                    "fleet saturated, keeping affinity with a queueing prefix owner \
+                     instead of diverting",
+                );
+            }
+        } else {
+            inputs.metrics.record_cache_aware_decision(
+                &inputs.model_id.0,
+                if cache_decision.queue_gate_fell_back {
+                    // The gate removed every owner and nowhere in the fleet is
+                    // unqueued, so the prefix was kept rather than traded for a
+                    // wait that cannot be dodged. Booked as saturation, never
+                    // as a plain hit.
+                    CacheAwareDecision::AllQueued
+                } else {
+                    CacheAwareDecision::CacheHit
+                },
+            );
+        }
         Some(decision.selected)
     }
 
@@ -525,6 +565,7 @@ fn prefill_policy_reason(
         PolicyKind::CacheAware => match (proposal, decision) {
             (_, DecisionReason::CacheCandidate)
             | (ProposalKind::CacheAffinity, DecisionReason::Primary) => "cache_candidate",
+            (_, DecisionReason::SaturationPin) => "saturation_pin",
             (_, DecisionReason::Primary) => "no_cache_candidate",
             (_, DecisionReason::BackupPrimaryAdmission) => "no_cache_candidate_admission_backup",
             (_, DecisionReason::BackupPressureGuard) => "no_cache_candidate_pressure_backup",
@@ -540,6 +581,7 @@ fn prefill_policy_reason(
             DecisionReason::BackupPressureGuard => "pressure_backup",
             DecisionReason::RangeFallback => "range_fallback",
             DecisionReason::CapacityFallbackPowerOfTwo => "capacity_fallback_power_of_two",
+            DecisionReason::SaturationPin => "saturation_pin",
         },
     }
 }
@@ -784,6 +826,7 @@ mod tests {
             load_snapshot,
             workers,
             worker_queue_limit: None,
+            saturation_queue_floor: None,
             ttft_slo_ms: None,
             tps_slo: None,
             session_affinity_mode: SessionAffinityMode::Bucket,

experimental/sgl-router/src/server/metrics.rs

@@ -66,15 +66,22 @@
 //!   `sgl_router_diverted_overlap_blocks` — read it against the overlap of
 //!   all selections: a diverted curve skewing high means the gate is trading
 //!   large cached prefixes for short waits.
-//! - `all_queued` — the queue gate removed every owner AND every worker in
-//!   the prefill fleet is queueing, so no diversion could dodge a wait. This
-//!   is the saturation signal for traffic the gate acted on, keyed on
-//!   saturation rather than on where the request landed: usually the request
-//!   kept its prefix, but when the re-admitted owners are also out of KV
-//!   capacity it lands off-owner and still books here. Reporting that case as `cache_miss` would hide the
-//!   saturation in the one state where it matters most. It deliberately
-//!   does NOT spell `cache_hit*`: a `decision=~"cache_hit.*"` hit-rate query
-//!   must not absorb it, or a fully saturated fleet reads as a healthy one.
+//! - `all_queued` — the queue gate removed every owner and no diversion could
+//!   dodge a wait. Two conditions draw it. Without `--saturation-queue-floor`
+//!   it means every worker in the prefill fleet is queueing at or above
+//!   `--worker-queue-limit`. With a floor set, the saturation pin also draws
+//!   it on the weaker condition the floor names: no fleet worker reads
+//!   strictly below the floor. Since the floor may be lower than the limit, a
+//!   floor well under the limit widens this label to fleets that still hold
+//!   gate-admissible workers — read it against the configured floor, not as
+//!   "every worker is over the limit". It is keyed on saturation rather than
+//!   on where the request landed: usually the request kept its prefix, but
+//!   when the owners it would keep are also out of KV capacity it lands
+//!   off-owner and still books here. Reporting that case as `cache_miss`
+//!   would hide the saturation in the one state where it matters most. It
+//!   deliberately does NOT spell `cache_hit*`: a `decision=~"cache_hit.*"`
+//!   hit-rate query must not absorb it, or a fully saturated fleet reads as a
+//!   healthy one.
 //!
 //! The four `sgl_router_worker*` gauges and `sgl_router_workers` are sampled
 //! at scrape time from the live [`crate::workers::WorkerRegistry`] (passed to

experimental/sgl-router/src/server/routes/chat.rs

@@ -636,6 +636,12 @@ pub async fn chat_completions(
         .affinity
         .as_ref()
         .and_then(|config| config.worker_queue_limit);
+    let saturation_queue_floor = ctx
+        .config
+        .model
+        .affinity
+        .as_ref()
+        .and_then(|config| config.saturation_queue_floor);
     // Each Bucket retry rebuilds the proposal and reruns Admission/Guard.
     let worker = select_prefill_worker(&PrefillSelectionInputs {
         policy: policy.as_ref(),
@@ -655,6 +661,7 @@ pub async fn chat_completions(
         tps_slo,
         session_affinity_mode,
         worker_queue_limit,
+        saturation_queue_floor,
     })
     .map_err(|reason| policy_selection_failed(&ctx, &model_str, reason))?;