refactor(router): move default worker selection into builtin plugin

Cargo.toml

@@ -55,6 +55,7 @@ repository = "https://github.com/ai-dynamo/dynamo.git"
 keywords = ["llm", "genai", "inference", "nvidia", "distributed"]
 
 [workspace.dependencies]
+dynamo-custom-policy-builtin = { path = "lib/router-plugins/builtin" }
 # Published separately by https://github.com/ai-dynamo/aisimulate.
 aisimulate-core = { version = "=0.12.0" }
 

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

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

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

@@ -268,9 +268,7 @@ mod tests {
         use axum::{Json, Router};
         use dynamo_kv_router::WorkerType;
         use dynamo_kv_router::config::kv_router_config_from_dynamo_env;
-        use dynamo_kv_router::services::selection::{
-            SelectionServiceBuilder, WorkerSelectionPolicyRegistry,
-        };
+        use dynamo_kv_router::services::selection::SelectionServiceBuilder;
         use kube::Config;
         use tokio::net::TcpListener;
         use tokio::sync::Notify;
@@ -285,7 +283,7 @@ mod tests {
             SelectionServiceBuilder::new(
                 kv_router_config_from_dynamo_env(),
                 WorkerType::Aggregated,
-                WorkerSelectionPolicyRegistry::default(),
+                dynamo_custom_policy_builtin::default_registry(),
             )
             .indexer_threads(1)
             .replica_sync(listener_port, Vec::new())

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

@@ -110,9 +110,12 @@ impl Selector {
             .await?;
         }
 
-        let mut builder =
-            SelectionServiceBuilder::new(kv_router_config, WorkerType::Aggregated, policy_registry)
-                .indexer_threads(cfg.selector_threads);
+        let mut builder = SelectionServiceBuilder::new(
+            kv_router_config,
+            WorkerType::Aggregated,
+            policy_registry.with_default_factory(dynamo_custom_policy_builtin::default_factory()),
+        )
+        .indexer_threads(cfg.selector_threads);
         if let Some(peer_replication) = peer_replication {
             builder = builder.replica_sync(peer_replication.sync_port, Vec::new());
         }

examples/router/custom-policy-example/README.md

@@ -20,7 +20,7 @@ policy crate -> catalog crate -> router-policy YAML -> frontend or EPP binary
 
 Dynamo owns discovery, eligibility, queueing, validation, reservations, accounting, and metrics. A policy sees only eligible workers and returns one candidate row.
 
-Preferred routing taints are optional candidate metadata. A filter, scorer, or picker must request `WorkerInputs::PREFERRED_TAINT` before reading `preferred_taint_multiplier()` from a candidate; otherwise, Dynamo does not materialize the multiplier. Exact hard-pinned requests also do not materialize it. Required routing taints remain Dynamo eligibility rules.
+Preferred routing taints are optional candidate metadata. A filter, scorer, or picker must request `WorkerInputs::PREFERRED_TAINT` before reading `preferred_taint_multiplier()` from a candidate; otherwise, that component receives `None`, even if another component requested it. Exact hard-pinned requests also do not materialize it. Required routing taints remain Dynamo eligibility rules.
 
 ## Pick a Starting Point
 
@@ -31,7 +31,7 @@ Preferred routing taints are optional candidate metadata. A filter, scorer, or p
 | `disagg-filter-score-pick` | Prefill and decode workers each need the complete policy flow |
 | `simple-stacked-score-pick` | Multiple scorer costs compose before one picker runs |
 
-The `simple-filter-score-pick` policy shows the complete pipeline. It filters on minimum device overlap and scores active requests. Its picker normally selects the lowest cost. Tool-result turns select the worker with the most device overlap through `session_context().input_trigger()`.
+The `simple-filter-score-pick` policy shows the complete pipeline. It filters on minimum device overlap, receives all surviving workers in one `score` call, and scores active requests above the least-loaded worker. A private preparation helper computes the minimum from that batch. The scorer writes one finite contribution per row into the host-owned output slice. Its picker normally selects the lowest cost. Tool-result turns select the worker with the most device overlap through `session_context().input_trigger()`.
 
 The [`soft-pin-repin` policy](soft-pin-repin/README.md) documents its load threshold, soft-binding behavior, and two-Mocker `A -> B -> B` walkthrough.
 
@@ -110,7 +110,7 @@ fn provider(
             config.clone(),
             worker_type.as_str(),
             filters,
-            vec![Box::new(ActiveRequestsScorer)],
+            vec![Box::new(ActiveRequestsScorer::default())],
             Box::new(RequestAwarePicker),
         )
     }))
@@ -372,3 +372,13 @@ python3 -m dynamo.mocker \
 ```
 
 Send the same `curl` request from a fourth terminal. The frontend log records separate prefill and decode selections. Each worker set runs its own filter, scorer, and picker.
+
+## Exact Prompt Length
+
+Filters, scorers, and pickers can read `context.prompt_tokens()` for the exact input token count. This returns the existing request value without a prompt copy or allocation. Use `context.request_blocks()` when the policy needs rounded KV blocks instead. For example, at 16 tokens per block, a 17-token prompt has 2 blocks.
+
+Request-level cache information also requires `WorkerInputs::CACHE`: `context.cache()` returns `None` without that declaration. With it, the borrowed view exposes `has_tier_matches()` and `shared_hits()`. The former is false when the snapshot has no tier matches; the latter is `None` when no shared-cache result was supplied. A policy can call `shared_hits().map(|hits| hits.hits_beyond(prefix))` to calculate shared hits after its chosen prefix, only when needed.
+
+Scorers receive a `WorkerCandidates` view with `iter()`, `get(row)`, `len()`, and `is_empty()`. Filters receive one borrowed `WorkerCandidate`. Both expose `cache()`, `load()`, and `preferred_taint_multiplier()` only when that component declares the corresponding `CACHE`, `LOAD`, or `PREFERRED_TAINT` group. The same rule applies to picker inputs. Another component requesting a group does not grant access. Views share the host's stored rows without allocating or copying a candidate table.
+
+The builtin default now uses the public policy API too. Rust hosts install `dynamo_custom_policy_builtin::default_registry()` before adding custom providers. This supplies default selection for roles that the custom configuration does not replace.

examples/router/custom-policy-example/disagg-filter-score-pick/src/filter.rs

@@ -19,7 +19,7 @@ impl WorkerFilter for MinimumDeviceOverlapFilter {
     fn keep(
         &mut self,
         _context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
+        candidate: WorkerCandidate<'_>,
     ) -> Result<bool, WorkerSelectionPolicyError> {
         let cache = candidate
             .cache()

examples/router/custom-policy-example/disagg-filter-score-pick/src/scorer.rs

@@ -4,7 +4,8 @@
 //! Load scorers for prefill and decode routing partitions.
 
 use dynamo_kv_router::{
-    WorkerCandidate, WorkerInputs, WorkerScorer, WorkerSelectionContext, WorkerSelectionPolicyError,
+    WorkerCandidates, WorkerInputs, WorkerScorer, WorkerSelectionContext,
+    WorkerSelectionPolicyError,
 };
 
 /// Scores prefill workers by active work and uncached request blocks.
@@ -18,19 +19,23 @@ impl WorkerScorer for PrefillLoadScorer {
     fn score(
         &mut self,
         context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
-    ) -> Result<f64, WorkerSelectionPolicyError> {
-        let load = candidate
-            .load()
-            .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
-        let cache = candidate
-            .cache()
-            .ok_or_else(|| WorkerSelectionPolicyError::failed("cache input unavailable"))?;
-        let active_prefill_blocks =
-            load.active_prefill_tokens() as f64 / context.block_size() as f64;
-        let uncached_request_blocks =
-            (context.request_blocks() as f64 - cache.device_overlap_blocks()).max(0.0);
-        Ok(active_prefill_blocks + uncached_request_blocks)
+        candidates: WorkerCandidates<'_>,
+        costs: &mut [f64],
+    ) -> Result<(), WorkerSelectionPolicyError> {
+        for (candidate, cost) in candidates.iter().zip(costs) {
+            let load = candidate
+                .load()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
+            let cache = candidate
+                .cache()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("cache input unavailable"))?;
+            let active_prefill_blocks =
+                load.active_prefill_tokens() as f64 / context.block_size() as f64;
+            let uncached_request_blocks =
+                (context.request_blocks() as f64 - cache.device_overlap_blocks()).max(0.0);
+            *cost = active_prefill_blocks + uncached_request_blocks;
+        }
+        Ok(())
     }
 }
 
@@ -47,11 +52,15 @@ impl WorkerScorer for DecodeLoadScorer {
     fn score(
         &mut self,
         _context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
-    ) -> Result<f64, WorkerSelectionPolicyError> {
-        let load = candidate
-            .load()
-            .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
-        Ok(load.decode_cost_blocks())
+        candidates: WorkerCandidates<'_>,
+        costs: &mut [f64],
+    ) -> Result<(), WorkerSelectionPolicyError> {
+        for (candidate, cost) in candidates.iter().zip(costs) {
+            let load = candidate
+                .load()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
+            *cost = load.decode_cost_blocks();
+        }
+        Ok(())
     }
 }

examples/router/custom-policy-example/simple-filter-score-pick/src/filter.rs

@@ -22,7 +22,7 @@ impl WorkerFilter for MinimumDeviceOverlapFilter {
     fn keep(
         &mut self,
         _context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
+        candidate: WorkerCandidate<'_>,
     ) -> Result<bool, WorkerSelectionPolicyError> {
         let cache = candidate
             .cache()

examples/router/custom-policy-example/simple-filter-score-pick/src/lib.rs

@@ -56,7 +56,7 @@ fn provider(
                 config.clone(),
                 worker_type.as_str(),
                 filters,
-                vec![Box::new(ActiveRequestsScorer)],
+                vec![Box::new(ActiveRequestsScorer::default())],
                 Box::new(RequestAwarePicker),
             )
         },

examples/router/custom-policy-example/simple-filter-score-pick/src/scorer.rs

@@ -4,27 +4,52 @@
 //! Active-request scorer for the `simple-filter-score-pick` policy.
 
 use dynamo_kv_router::plugins::worker_selection::{
-    WorkerCandidate, WorkerInputs, WorkerScorer, WorkerSelectionContext, WorkerSelectionPolicyError,
+    WorkerCandidates, WorkerInputs, WorkerScorer, WorkerSelectionContext,
+    WorkerSelectionPolicyError,
 };
 
-/// Scores each candidate by its current number of active requests.
-pub(crate) struct ActiveRequestsScorer;
+/// Scores active requests above the least-loaded surviving candidate.
+#[derive(Default)]
+pub(crate) struct ActiveRequestsScorer {
+    minimum: usize,
+}
+impl ActiveRequestsScorer {
+    fn prepare(
+        &mut self,
+        _context: &WorkerSelectionContext<'_>,
+        candidates: WorkerCandidates<'_>,
+    ) -> Result<(), WorkerSelectionPolicyError> {
+        self.minimum = usize::MAX;
+        for candidate in candidates.iter() {
+            let load = candidate
+                .load()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
+            self.minimum = self.minimum.min(load.active_requests());
+        }
+        Ok(())
+    }
+}
 
 impl WorkerScorer for ActiveRequestsScorer {
     /// Requests load inputs for the active-request count.
     fn required_worker_inputs(&self) -> WorkerInputs {
         WorkerInputs::LOAD
     }
 
-    /// Returns the active-request count as a lower-is-better cost.
+    /// Prepare the batch minimum and write one relative load cost per worker.
     fn score(
         &mut self,
-        _context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
-    ) -> Result<f64, WorkerSelectionPolicyError> {
-        let load = candidate
-            .load()
-            .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
-        Ok(load.active_requests() as f64)
+        context: &WorkerSelectionContext<'_>,
+        candidates: WorkerCandidates<'_>,
+        costs: &mut [f64],
+    ) -> Result<(), WorkerSelectionPolicyError> {
+        self.prepare(context, candidates)?;
+        for (candidate, cost) in candidates.iter().zip(costs) {
+            let load = candidate
+                .load()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
+            *cost = (load.active_requests() - self.minimum) as f64;
+        }
+        Ok(())
     }
 }

examples/router/custom-policy-example/simple-stacked-score-pick/src/scorer/active_requests.rs

@@ -4,7 +4,8 @@
 //! Active-request cost for the stacked scorer stage.
 
 use dynamo_kv_router::{
-    WorkerCandidate, WorkerInputs, WorkerScorer, WorkerSelectionContext, WorkerSelectionPolicyError,
+    WorkerCandidates, WorkerInputs, WorkerScorer, WorkerSelectionContext,
+    WorkerSelectionPolicyError,
 };
 
 pub(crate) struct ActiveRequestsScorer;
@@ -17,11 +18,15 @@ impl WorkerScorer for ActiveRequestsScorer {
     fn score(
         &mut self,
         _context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
-    ) -> Result<f64, WorkerSelectionPolicyError> {
-        let load = candidate
-            .load()
-            .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
-        Ok(load.active_requests() as f64)
+        candidates: WorkerCandidates<'_>,
+        costs: &mut [f64],
+    ) -> Result<(), WorkerSelectionPolicyError> {
+        for (candidate, cost) in candidates.iter().zip(costs) {
+            let load = candidate
+                .load()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("load input unavailable"))?;
+            *cost = load.active_requests() as f64;
+        }
+        Ok(())
     }
 }

examples/router/custom-policy-example/simple-stacked-score-pick/src/scorer/uncached_blocks.rs

@@ -4,7 +4,8 @@
 //! Uncached-block cost for the stacked scorer stage.
 
 use dynamo_kv_router::{
-    WorkerCandidate, WorkerInputs, WorkerScorer, WorkerSelectionContext, WorkerSelectionPolicyError,
+    WorkerCandidates, WorkerInputs, WorkerScorer, WorkerSelectionContext,
+    WorkerSelectionPolicyError,
 };
 
 pub(crate) struct UncachedBlocksScorer;
@@ -17,11 +18,15 @@ impl WorkerScorer for UncachedBlocksScorer {
     fn score(
         &mut self,
         context: &WorkerSelectionContext<'_>,
-        candidate: &WorkerCandidate,
-    ) -> Result<f64, WorkerSelectionPolicyError> {
-        let cache = candidate
-            .cache()
-            .ok_or_else(|| WorkerSelectionPolicyError::failed("cache input unavailable"))?;
-        Ok((context.request_blocks() as f64 - cache.device_overlap_blocks()).max(0.0))
+        candidates: WorkerCandidates<'_>,
+        costs: &mut [f64],
+    ) -> Result<(), WorkerSelectionPolicyError> {
+        for (candidate, cost) in candidates.iter().zip(costs) {
+            let cache = candidate
+                .cache()
+                .ok_or_else(|| WorkerSelectionPolicyError::failed("cache input unavailable"))?;
+            *cost = (context.request_blocks() as f64 - cache.device_overlap_blocks()).max(0.0);
+        }
+        Ok(())
     }
 }

lib/bench/kv_router/dc_ckf_relay_bench.rs

@@ -992,7 +992,7 @@ async fn drive_phase(
     tokio::task::spawn_blocking(move || {
         let ready = Arc::new(Barrier::new(issuer_count + 1));
         let start = Arc::new(Barrier::new(issuer_count + 1));
-        let started = Arc::new(OnceLock::new());
+        let started = Arc::new(OnceLock::<Instant>::new());
         let (elapsed, outputs) = std::thread::scope(|scope| {
             let mut handles = Vec::with_capacity(issuer_count);
             for (issuer, shard) in shards.into_iter().enumerate() {