feat(router): add pluggable hint policies to the router

components/src/dynamo/common/configuration/groups/kv_router_args.py

@@ -241,6 +241,22 @@ def _router_settings(self) -> dict[str, Any]:
         except ValidationError as exc:
             raise ValueError(f"invalid router settings: {exc}") from exc
 
+    def hint_policy_roles(self) -> list[str]:
+        """Roles that list hint policies in --router-policy-config.
+
+        A role listed with an empty list opts out of the defaults and runs none.
+        """
+        if self.router_policy_config is None:
+            return []
+        with Path(self.router_policy_config).open() as source:
+            document = yaml.safe_load(source)
+        hint_policies = (
+            document.get("hint_policies") if isinstance(document, dict) else None
+        )
+        if not isinstance(hint_policies, dict):
+            return []
+        return [role for role, entries in hint_policies.items() if entries]
+
     def apply_router_config(self) -> None:
         self.apply_load_aware_preset()
         self.apply_conditional_disagg_config()
@@ -627,7 +643,7 @@ def add_arguments(self, parser) -> None:
             default=None,
             help=(
                 "KV Router: Startup-only YAML configuration for shared router settings, "
-                "policy-class queues, and worker-selection instances. "
+                "policy-class queues, worker-selection instances, and hint policies. "
                 "When omitted, router_queue_threshold and router_queue_policy define "
                 "one synthetic policy class; queueing remains disabled unless "
                 "router_queue_threshold is set."

components/src/dynamo/common/configuration/groups/router_args.py

@@ -434,6 +434,9 @@ def router_kwargs(self) -> dict:
     def kv_router_kwargs(self) -> dict:
         ...
 
+    def hint_policy_roles(self) -> list[str]:
+        ...
+
 
 def build_router_config(
     config: Optional[RouterConfigSource],
@@ -474,6 +477,13 @@ def build_router_config(
         ) from error
 
     mode = getattr(RouterMode, mode_attr)
+    if mode != RouterMode.KV and (roles := config.hint_policy_roles()):
+        # Hint policies run when the KV path books a worker; other modes would drop them.
+        raise ValueError(
+            f"hint_policies for {', '.join(roles)} run only on the KV path, but "
+            f"--router-mode {router_mode} routes without it; use --router-mode kv, "
+            "or remove those roles' hint_policies from --router-policy-config"
+        )
     # Only KV routing consults KvRouterConfig; passing it for other modes would
     # imply tuning that is never read.
     kv_router_config = (

components/src/dynamo/common/tests/configuration/test_router_advertisement.py

@@ -145,6 +145,31 @@ def test_unknown_mode_is_rejected():
         build_router_config(config)
 
 
+def _policy_config(tmp_path, hint_policies: str) -> str:
+    path = tmp_path / "router-policy.yaml"
+    path.write_text(f"hint_policies:\n{hint_policies}")
+    return str(path)
+
+
+@pytest.mark.parametrize("mode", ["round-robin", "least-loaded", "direct"])
+def test_hint_policies_off_the_kv_path_are_rejected(tmp_path, mode):
+    """Only the KV path runs hint policies, so another mode would drop them silently."""
+    policy = _policy_config(tmp_path, "  aggregated:\n    - type: kv_fetch\n")
+    config, _ = parse_worker_router_config(
+        ["--router-mode", mode, "--router-policy-config", policy]
+    )
+    with pytest.raises(ValueError, match="hint_policies for aggregated run only"):
+        build_router_config(config)
+
+
+def test_hint_policy_opt_out_is_allowed_off_the_kv_path(tmp_path):
+    policy = _policy_config(tmp_path, "  aggregated: []\n")
+    config, _ = parse_worker_router_config(
+        ["--router-mode", "round-robin", "--router-policy-config", policy]
+    )
+    assert build_router_config(config) is not None
+
+
 @pytest.mark.parametrize("backend", ["vllm", "trtllm", "sglang"])
 def test_router_flags_do_not_collide_with_backend_flags(backend):
     """Guard against a backend later adding a flag the router group owns.

docs/fern/pages/developer-guide/knowledge-base/modular-components/router/configuration-and-tuning.md

@@ -79,7 +79,7 @@ its configuration even if the endpoint is unchanged.
 - `--router-prefill-load-model`: Selects the router's prompt-side load model. `none` keeps the existing static prompt load accounting. `ais` predicts one expected prefill duration per admitted request and lazily decays only the oldest active prefill request on each worker.
 - `--router-queue-threshold`: **Deprecated; use `policy_classes[].prefill_busy_threshold_frac` in the policy YAML.** Optional queue threshold fraction for prefill token capacity. Queueing is disabled by default; setting a numeric value enables it. The router holds incoming requests in a priority queue while all eligible workers exceed `threshold * max_num_batched_tokens`, releasing them when capacity frees up. This defers dispatch rather than rejecting work, so routing decisions use the freshest load metrics at the moment a request is sent to a worker. `nvext.agent_hints.strict_priority` selects an absolute pending-queue tier, while `nvext.agent_hints.priority` adjusts ordering within the configured policy. Must be greater than or equal to 0; use `0.0` for maximum queueing sensitivity. See the SGLang note under [Tuning Guidelines](#tuning-guidelines) for caveats around how `max_num_batched_tokens` is populated on that backend, and see [Priority Scheduling](../../../../use-cases/agents/priority-scheduling.md) for how router priority differs from backend engine priority.
 - `--router-queue-policy`: **Deprecated; use `policy_classes[].queue_policy` in the policy YAML.** Scheduling policy for the router queue: `fcfs` (default) or `wspt`.
-- `--router-policy-config`: Startup-only YAML path for shared router settings, policy-class queues, and worker-selection instances. When omitted, `--router-queue-threshold` and `--router-queue-policy` define one synthetic policy class. The equivalent environment variable is `DYN_ROUTER_POLICY_CONFIG`. See [Worker-Selection Policies](#worker-selection-policies) to select a built-in policy, and [Write Custom Routing Strategies](custom-worker-selection.mdx) for the linked-policy schema.
+- `--router-policy-config`: Startup-only YAML path for shared router settings, policy-class queues, worker-selection instances, and hint policies. When omitted, `--router-queue-threshold` and `--router-queue-policy` define one synthetic policy class. The equivalent environment variable is `DYN_ROUTER_POLICY_CONFIG`. See [Worker-Selection Policies](#worker-selection-policies) to select a built-in policy, and [Write Custom Routing Strategies](custom-worker-selection.mdx) for the linked-policy schema.
 
 For how queue backpressure differs from candidate filtering and busy-threshold overload handling, see [Router Filtering](worker-filtering.md).
 
@@ -264,6 +264,31 @@ for that scope, which makes it a quick way to A/B a policy against the default.
 To write your own policy instead of using a built-in one, see
 [Write Custom Routing Strategies](custom-worker-selection.mdx).
 
+### Hint Policies
+
+Hint policies are experimental. They add engine-directed KV actions, such as a [KV fetch hint](topology-aware-kv-transfer.md#kv-fetch-hints), to the worker a booking selected. They never change which worker is selected. Hint policies are plugin types that a linked plugin crate registers.
+
+The `hint_policies` section of the same YAML file lists, for each worker role, the hint policies that role runs, in order. Each entry has a `type` and optional `parameters`. Add a `name` only when one role lists the same type twice:
+
+```yaml
+hint_policies:
+  decode:
+    - type: harness_prefetch
+      name: short
+      parameters:
+        max_blocks: 64
+    - type: harness_prefetch
+      name: long
+      parameters:
+        max_blocks: 256
+  prefill: []
+```
+
+- **A role that `hint_policies` doesn't list runs the default hint policies,** if the router image supplies any. Listing the role, even as an empty list, replaces them. Default hint policies are passive: they start no resources and don't change how the role is routed.
+- **Configured hint policies run only on the KV path,** because the router composes hints when it books a worker. Startup fails when a role with configured hint policies runs under a non-KV `--router-mode`. To use a routing mode's policy with hint policies, run `--router-mode kv` and name the routing type, such as `round-robin`, under `worker_selection` for that role.
+- **The router composes one hint per booking** and numbers its actions `a1`, `a2`, and so on. It sends the hint only to workers that advertise `router_hint`. A policy that fails drops only its own actions. A hint carries at most 32 actions.
+- **At most one hint policy per role may send the KV copies a worker-selection policy commits to.** Startup fails when two do. A hint policy that reads KV copy sources needs a local, event-driven KV index; startup fails without one.
+
 ### Policy-Class Queues
 
 YAML profiles define a matrix from client-requested policy family and

lib/kv-router/src/indexer/types.rs

@@ -448,7 +448,7 @@ pub struct MatchDetails {
     pub overlap_scores: OverlapScores,
     /// Last matched device sequence hash per worker, used to seed lower-tier queries.
     pub last_matched_hashes: FxHashMap<WorkerWithDpRank, ExternalSequenceBlockHash>,
-    /// Optional root-aligned device candidates used to build compact KV transfer hints.
+    /// Optional root-aligned device candidates that KV copy sources are found from.
     pub kv_transfer_candidates: Option<KvTransferCandidates>,
 }
 
@@ -524,7 +524,7 @@ pub struct MatchDetailsRequest {
     pub sequence: Vec<LocalBlockHash>,
     /// A boolean indicating whether to exit early if a single match is found.
     pub early_exit: bool,
-    /// When true, retain the matched root-aligned external hash chain for KV transfer hints.
+    /// When true, retain the matched root-aligned external hash chain for KV copy sources.
     pub retain_kv_transfer_chain: bool,
     /// A channel sender to send the `MatchDetails` response.
     pub resp: oneshot::Sender<MatchDetails>,

lib/kv-router/src/kv_hints.rs

@@ -13,9 +13,12 @@ use crate::protocols::{
     ExternalSequenceBlockHash, ResidencyOwnerKey, ResidencyRoutingSnapshot, WorkerWithDpRank,
 };
 
+#[cfg_attr(not(feature = "standalone-selection"), allow(unused_imports))]
+pub(crate) use sources::kv_copy_locations;
 pub use sources::{
-    CacheOwnerKvTransferKeys, KvTransferScope, KvTransferSourceChoice, KvTransferSourceSummary,
-    KvTransferTarget, WorkerKvTransferKeys, WorkerKvTransferKeysCache,
+    CacheOwnerKvTransferKeys, KvCopyLocations, KvCopySource, KvTransferScope,
+    KvTransferSourceChoice, KvTransferSourceSummary, KvTransferTarget, WorkerKvTransferKeys,
+    WorkerKvTransferKeysCache,
 };
 
 /// The selected worker can consume a `TRANSFER` hint with the v1 payload.
@@ -28,6 +31,8 @@ pub const KV_HINT_TRANSFER_SOURCE_CONTROL_ENDPOINTS_RUNTIME_KEY: &str =
     "router_hint_source_control_endpoints";
 
 const KV_HINT_PROTOCOL_VERSION: &str = "0.1";
+/// Upper bound on the actions one hint carries.
+pub const MAX_KV_HINT_ACTIONS: usize = 32;
 const KV_FETCH_ACTION_TYPE: &str = "kv.fetch";
 const KV_FETCH_ACTION_VERSION: &str = "1.0";
 
@@ -53,6 +58,8 @@ pub struct KvHintAction {
 }
 
 impl KvHintAction {
+    /// An action with an explicit ID. When the router composes a hint, it numbers every action
+    /// again (see [`KvHint::append_numbered`]), so a hint policy can use [`Self::unnumbered`].
     pub fn new(
         action_id: impl Into<String>,
         action_type: impl Into<String>,
@@ -67,6 +74,15 @@ impl KvHintAction {
         }
     }
 
+    /// An action without an ID. The router numbers it when it composes the hint.
+    pub fn unnumbered(
+        action_type: impl Into<String>,
+        action_version: impl Into<String>,
+        payload: BTreeMap<String, serde_json::Value>,
+    ) -> Self {
+        Self::new(String::new(), action_type, action_version, payload)
+    }
+
     pub fn fetch(action_id: impl Into<String>, payload: KvSourceLocationsPayload) -> Self {
         let KvSourceLocationsPayload {
             source_control_endpoint,
@@ -103,6 +119,50 @@ impl KvHint {
             actions,
         }
     }
+
+    /// An empty hint for one routed request. Its message ID is the request ID.
+    pub fn for_request(request_id: impl Into<String>) -> Self {
+        Self::new(request_id, Vec::new())
+    }
+
+    /// Append one producer's actions whole, numbering each by its position in the hint: `a1`,
+    /// `a2`, and so on. Numbering replaces any ID the producer set, so action IDs are unique
+    /// without producers coordinating them.
+    ///
+    /// Rejects all of `actions`, leaving the hint unchanged, when they would take it past
+    /// [`MAX_KV_HINT_ACTIONS`].
+    pub fn append_numbered(
+        &mut self,
+        actions: Vec<KvHintAction>,
+    ) -> Result<(), KvHintActionLimitError> {
+        if self.actions.len() + actions.len() > MAX_KV_HINT_ACTIONS {
+            return Err(KvHintActionLimitError {
+                present: self.actions.len(),
+                added: actions.len(),
+            });
+        }
+        for mut action in actions {
+            action.action_id = format!("a{}", self.actions.len() + 1);
+            self.actions.push(action);
+        }
+        Ok(())
+    }
+
+    /// Whether the hint carries no action.
+    pub fn is_empty(&self) -> bool {
+        self.actions.is_empty()
+    }
+}
+
+/// Actions that would take a hint past [`MAX_KV_HINT_ACTIONS`].
+#[derive(Debug, thiserror::Error)]
+#[error(
+    "a hint carries at most {max} actions; it has {present} and {added} more were offered",
+    max = MAX_KV_HINT_ACTIONS
+)]
+pub struct KvHintActionLimitError {
+    present: usize,
+    added: usize,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
@@ -174,6 +234,37 @@ mod tests {
         );
     }
 
+    #[test]
+    fn appended_actions_are_numbered_and_bounded() {
+        let action = |id: &str| KvHintAction::new(id, "test.action", "1", BTreeMap::new());
+        let mut hint = KvHint::for_request("request-1");
+        assert!(hint.is_empty());
+        hint.append_numbered(vec![
+            action("x"),
+            KvHintAction::unnumbered("test.action", "1", BTreeMap::new()),
+        ])
+        .unwrap();
+        hint.append_numbered(vec![action("x")]).unwrap();
+        let ids: Vec<_> = hint
+            .actions
+            .iter()
+            .map(|action| action.action_id.as_str())
+            .collect();
+        assert_eq!(ids, ["a1", "a2", "a3"]);
+        assert_eq!(hint.message_id, "request-1");
+
+        // A batch that would overflow the hint is rejected whole.
+        let error = hint
+            .append_numbered(vec![action("y"); MAX_KV_HINT_ACTIONS - 2])
+            .unwrap_err();
+        assert!(error.to_string().contains("at most 32 actions"), "{error}");
+        assert_eq!(hint.actions.len(), 3);
+        hint.append_numbered(vec![action("y"); MAX_KV_HINT_ACTIONS - 3])
+            .unwrap();
+        assert_eq!(hint.actions.len(), MAX_KV_HINT_ACTIONS);
+        assert_eq!(hint.actions[MAX_KV_HINT_ACTIONS - 1].action_id, "a32");
+    }
+
     #[test]
     fn deserializes_unknown_action_as_opaque_transport_data() {
         let value = serde_json::json!({