feat(frontend): add exact MM routing for generate

components/src/dynamo/vllm/engine_generate.py

@@ -8,15 +8,17 @@
 from dynamo.llm import ModelInput, ModelRuntimeConfig, ModelType, WorkerType
 
 VLLM_GENERATE_CAPABILITY = "vllm_inference_v1_generate"
+VLLM_ENABLE_TOWER_CONNECTOR_LORA_RUNTIME_KEY = "vllm_enable_tower_connector_lora"
 
 
 def publish_engine_generate_capability(
     runtime_config: ModelRuntimeConfig,
     model_input: ModelInput,
     model_type: ModelType,
     worker_type: WorkerType,
+    tower_connector_lora_enabled: bool,
 ) -> bool:
-    """Publish native Generate support when the worker accepts token input."""
+    """Publish native Generate support and its MM-routing-relevant config."""
     if model_input != ModelInput.Tokens:
         return False
     if worker_type == WorkerType.Prefill:
@@ -32,4 +34,8 @@ def publish_engine_generate_capability(
         VLLM_GENERATE_CAPABILITY,
         json.dumps(True),
     )
+    runtime_config.set_engine_specific(
+        VLLM_ENABLE_TOWER_CONNECTOR_LORA_RUNTIME_KEY,
+        json.dumps(tower_connector_lora_enabled),
+    )
     return True

components/src/dynamo/vllm/main.py

@@ -744,8 +744,16 @@ async def register_vllm_model(
         runtime_config, config.engine_args, worker_type, dp_range
     )
     runtime_config.context_length = vllm_config.model_config.max_model_len
+    tower_connector_lora_enabled = bool(
+        vllm_config.lora_config
+        and getattr(vllm_config.lora_config, "enable_tower_connector_lora", False)
+    )
     if publish_engine_generate_capability(
-        runtime_config, model_input, model_type, worker_type
+        runtime_config,
+        model_input,
+        model_type,
+        worker_type,
+        tower_connector_lora_enabled,
     ):
         logging.info("Published vLLM engine-native generate capability")
     if model_type != ModelType.Embedding:

components/src/dynamo/vllm/tests/test_runtime_metadata.py

@@ -3,14 +3,15 @@
 
 import json
 from types import SimpleNamespace
-from unittest.mock import Mock
+from unittest.mock import Mock, call
 
 import pytest
 
 from dynamo.common.token_budget import TOKEN_BUDGET_RUNTIME_KEY
 from dynamo.llm import ModelInput, ModelType, WorkerType
 from dynamo.vllm.capacity import get_metrics_model_name, get_spec_decode_runtime_data
 from dynamo.vllm.engine_generate import (
+    VLLM_ENABLE_TOWER_CONNECTOR_LORA_RUNTIME_KEY,
     VLLM_GENERATE_CAPABILITY,
     publish_engine_generate_capability,
 )
@@ -69,31 +70,62 @@ def test_vllm_token_budget_matches_rejection_policy():
 
 
 @pytest.mark.parametrize(
-    ("model_input", "model_type", "worker_type", "expected"),
+    (
+        "model_input",
+        "model_type",
+        "worker_type",
+        "tower_connector_lora_enabled",
+        "expected",
+    ),
     [
-        (ModelInput.Tokens, ModelType.Prefill, WorkerType.Prefill, True),
-        (ModelInput.Tokens, ModelType.Chat, WorkerType.Decode, True),
-        (ModelInput.Tokens, ModelType.Completions, WorkerType.Aggregated, True),
-        (ModelInput.Tokens, ModelType.Empty, WorkerType.Prefill, False),
-        (ModelInput.Tokens, ModelType.Empty, WorkerType.Decode, False),
-        (ModelInput.Text, ModelType.Chat, WorkerType.Aggregated, False),
-        (ModelInput.Tokens, ModelType.Embedding, WorkerType.Aggregated, False),
+        (ModelInput.Tokens, ModelType.Prefill, WorkerType.Prefill, False, True),
+        (ModelInput.Tokens, ModelType.Chat, WorkerType.Decode, True, True),
+        (
+            ModelInput.Tokens,
+            ModelType.Completions,
+            WorkerType.Aggregated,
+            False,
+            True,
+        ),
+        (ModelInput.Tokens, ModelType.Empty, WorkerType.Prefill, False, False),
+        (ModelInput.Tokens, ModelType.Empty, WorkerType.Decode, False, False),
+        (ModelInput.Text, ModelType.Chat, WorkerType.Aggregated, True, False),
+        (
+            ModelInput.Tokens,
+            ModelType.Embedding,
+            WorkerType.Aggregated,
+            False,
+            False,
+        ),
     ],
 )
 def test_vllm_generate_capability_publication(
-    model_input, model_type, worker_type, expected
+    model_input,
+    model_type,
+    worker_type,
+    tower_connector_lora_enabled,
+    expected,
 ):
     runtime_config = SimpleNamespace(set_engine_specific=Mock())
 
     published = publish_engine_generate_capability(
-        runtime_config, model_input, model_type, worker_type
+        runtime_config,
+        model_input,
+        model_type,
+        worker_type,
+        tower_connector_lora_enabled,
     )
 
     assert published is expected
     if expected:
-        runtime_config.set_engine_specific.assert_called_once_with(
-            VLLM_GENERATE_CAPABILITY, json.dumps(True)
-        )
+        expected_calls = [
+            call(VLLM_GENERATE_CAPABILITY, json.dumps(True)),
+            call(
+                VLLM_ENABLE_TOWER_CONNECTOR_LORA_RUNTIME_KEY,
+                json.dumps(tower_connector_lora_enabled),
+            ),
+        ]
+        assert runtime_config.set_engine_specific.call_args_list == expected_calls
     else:
         runtime_config.set_engine_specific.assert_not_called()
 

lib/kv-router/src/protocols.rs

@@ -79,6 +79,20 @@ pub fn pad_value_for_mm_hash(mm_hash: u64) -> u32 {
     (MM_PAD_SHIFT_VALUE + (mm_hash & MM_PAD_HASH_MASK)) as u32
 }
 
+/// Map a non-empty multimodal identifier to Dynamo's routing hash.
+///
+/// Preserve vLLM's canonical 64-character hex-digest mapping for compatibility,
+/// and hash shorter opaque identifiers emitted by external renderers with XXH3.
+pub fn hash_mm_identifier(identifier: &str) -> Option<u64> {
+    if identifier.is_empty() {
+        return None;
+    }
+    if identifier.len() == 64 && identifier.chars().all(|c| c.is_ascii_hexdigit()) {
+        return u64::from_str_radix(&identifier[..16], 16).ok();
+    }
+    Some(xxh3::xxh3_64(identifier.as_bytes()))
+}
+
 /// Compute the hash for a sequence of tokens, optionally including multimodal metadata,
 /// LoRA adapter identity, and cache namespace.
 ///
@@ -1870,6 +1884,22 @@ mod tests {
         );
     }
 
+    #[test]
+    fn mm_identifier_hash_preserves_canonical_vllm_digest_mapping() {
+        let identifier = "0123456789abcdef".repeat(4);
+        assert_eq!(hash_mm_identifier(&identifier), Some(0x0123_4567_89ab_cdef));
+    }
+
+    #[test]
+    fn mm_identifier_hash_supports_opaque_identifiers() {
+        let identifier = "opaque-renderer-image-0";
+        assert_eq!(
+            hash_mm_identifier(identifier),
+            Some(xxh3::xxh3_64(identifier.as_bytes()))
+        );
+        assert_eq!(hash_mm_identifier(""), None);
+    }
+
     #[test]
     fn test_router_event_new() {
         let worker_id = 0;

lib/kv-router/src/zmq_wire/convert.rs

@@ -167,14 +167,20 @@ pub fn convert_event(
 }
 
 /// Rewrite each `image_token_id` run in `token_ids` to `pad_value(mm_hash)`,
-/// one mm_hash per run in order, so the recomputed `tokens_hash` matches the
-/// frontend's pad_value expansion. Exact when images are separated by a
-/// non-image token (true for Qwen2/2.5/3-VL); a run-vs-mm_object count mismatch
-/// (adjacent images, no separator) is logged below rather than silent.
-fn substitute_pad_values(token_ids: &[u32], image_token_id: u32, mm_objects: &[u64]) -> Vec<u32> {
+/// assigning one MM hash per run in order and clamping excess runs to the last
+/// hash. Returns the normalized tokens and the number of discovered runs.
+///
+/// This is the shared request/event normalization contract. Callers decide
+/// whether a run/object count mismatch is acceptable for their use case.
+pub fn normalize_mm_token_runs(
+    token_ids: &[u32],
+    image_token_id: u32,
+    mm_hashes: &[u64],
+) -> Option<(Vec<u32>, usize)> {
+    let last_mm_hash = *mm_hashes.last()?;
     let mut out = Vec::with_capacity(token_ids.len());
     // `obj_idx` advances once per completed run, so run N fills with
-    // mm_objects[N], clamped to the last object if runs outnumber mm_objects.
+    // mm_hashes[N], clamped to the last object if runs outnumber hashes.
     let mut obj_idx = 0usize;
     let mut in_run = false;
     let mut runs = 0usize;
@@ -186,13 +192,7 @@ fn substitute_pad_values(token_ids: &[u32], image_token_id: u32, mm_objects: &[u
             if !in_run {
                 in_run = true;
                 runs += 1;
-                // Safety: the sole caller (`create_stored_block_from_parts`)
-                // only reaches here with a non-empty `mm_objects`, so `last()`
-                // is `Some`.
-                let mm_hash = mm_objects
-                    .get(obj_idx)
-                    .copied()
-                    .unwrap_or_else(|| *mm_objects.last().unwrap());
+                let mm_hash = mm_hashes.get(obj_idx).copied().unwrap_or(last_mm_hash);
                 run_pad = crate::protocols::pad_value_for_mm_hash(mm_hash);
             }
             out.push(run_pad);
@@ -204,14 +204,7 @@ fn substitute_pad_values(token_ids: &[u32], image_token_id: u32, mm_objects: &[u
             out.push(t);
         }
     }
-    if runs != mm_objects.len() {
-        tracing::debug!(
-            runs,
-            mm_objects = mm_objects.len(),
-            "image_token_id run count != mm_object count; pad_value assignment is best-effort by run order"
-        );
-    }
-    out
+    Some((out, runs))
 }
 
 #[derive(Default)]
@@ -246,7 +239,15 @@ pub fn create_stored_block_from_parts(
     let tokens_hash = match (image_token_id, mm_extra_info.as_ref()) {
         (Some(img_tok), Some(info)) if !info.mm_objects.is_empty() => {
             let mm_hashes: Vec<u64> = info.mm_objects.iter().map(|o| o.mm_hash).collect();
-            let substituted = substitute_pad_values(token_ids, img_tok, &mm_hashes);
+            let (substituted, runs) = normalize_mm_token_runs(token_ids, img_tok, &mm_hashes)
+                .expect("non-empty multimodal objects must normalize");
+            if runs != mm_hashes.len() {
+                tracing::debug!(
+                    runs,
+                    mm_objects = mm_hashes.len(),
+                    "image_token_id run count != mm_object count; pad_value assignment is best-effort by run order"
+                );
+            }
             compute_block_hash_for_seq(
                 &substituted,
                 kv_block_size,
@@ -360,6 +361,28 @@ mod normalize_tests {
     use super::*;
     use crate::protocols::{BlockMmObjectInfo, pad_value_for_mm_hash};
 
+    #[test]
+    fn mm_token_run_normalization_uses_worker_order_and_clamps_excess_runs() {
+        let image_token_id = 99;
+        let (normalized, runs) =
+            normalize_mm_token_runs(&[10, 99, 42, 99, 20, 99], image_token_id, &[7, 8])
+                .expect("non-empty hashes normalize");
+
+        assert_eq!(runs, 3);
+        assert_eq!(
+            normalized,
+            vec![
+                10,
+                pad_value_for_mm_hash(7),
+                42,
+                pad_value_for_mm_hash(8),
+                20,
+                pad_value_for_mm_hash(8),
+            ]
+        );
+        assert!(normalize_mm_token_runs(&[99], image_token_id, &[]).is_none());
+    }
+
     /// A normalized vLLM block (image_token_id run + mm_hash) must hash
     /// identically to the frontend's pad_value scheme. The parity the
     /// consolidation rests on.

lib/kv-router/src/zmq_wire/extra_keys.rs

@@ -8,6 +8,12 @@ use super::types::ExtraKeyItem;
 // Must match _DYNAMO_CACHE_SALT_PREFIX in components/src/dynamo/vllm/handlers.py.
 const DYNAMO_CACHE_SALT_PREFIX: &str = "dynamo-cache-salt:";
 
+/// Encode a frontend-approved routing hash in the marker form carried through
+/// vLLM `extra_keys`.
+pub fn mark_mm_hash_for_extra_key(mm_hash: u64) -> String {
+    format!("{mm_hash:016x}{}", "0".repeat(48))
+}
+
 /// Parse a frontend-issued Dynamo MM hash from a vLLM extra key.
 ///
 /// The worker canonicalizes a frontend-approved routing hash to its first 16
@@ -152,6 +158,10 @@ mod tests {
 
     #[test]
     fn only_frontend_padded_mm_hashes_are_parsed() {
+        assert_eq!(
+            parse_mm_hash_from_extra_key(&mark_mm_hash_for_extra_key(0x0123_4567_89ab_cdef)),
+            Some(0x0123_4567_89ab_cdef)
+        );
         assert_eq!(
             parse_mm_hash_from_extra_key(
                 "0123456789abcdef000000000000000000000000000000000000000000000000"

lib/kv-router/src/zmq_wire/mod.rs

@@ -25,9 +25,11 @@ mod types;
 
 pub use convert::{
     StoredBlockOptions, convert_event, create_stored_block_from_parts, create_stored_blocks,
+    normalize_mm_token_runs,
 };
 pub use extra_keys::{
-    extra_keys_to_block_mm_infos, extra_keys_to_cache_namespace, parse_mm_hash_from_extra_key,
+    extra_keys_to_block_mm_infos, extra_keys_to_cache_namespace, mark_mm_hash_for_extra_key,
+    parse_mm_hash_from_extra_key,
 };
 pub use filter::KvCacheSpecKind;
 pub use types::{

lib/llm/src/discovery.rs

@@ -3,6 +3,7 @@
 
 mod model;
 pub(crate) mod readiness;
+pub(crate) use model::GenerateEngineSelection;
 pub use model::Model;
 
 pub mod kv_source_membership;