feat(mm-routing): add Qwen video-aware KV routing for SGLang

components/src/dynamo/sglang/register.py

@@ -41,6 +41,9 @@
     runtime_capacity,
 )
 from dynamo.sglang.engine_generate import SGLANG_GENERATE_CAPABILITY
+from dynamo.sglang.video_routing import (
+    publish_sglang_qwen_video_processor_contract,
+)
 
 SGLANG_HICACHE_MOONCAKE_RUNTIME_KEY = "sglang_hicache_mooncake"
 SPEC_DECODE_RUNTIME_KEY = "spec_decode"
@@ -405,6 +408,15 @@ async def get_runtime_config(
     # generation overflow handling to their downstream backend.
     if engine is not None:
         publish_token_budget(runtime_config, _get_token_budget(engine, server_args))
+        # Hash forwarding currently exists on SGLang's aggregated generation
+        # path. Do not advertise exact video routing to disaggregated workers,
+        # whose prefill/decode handlers would otherwise publish incompatible
+        # KV-event placeholder hashes.
+        if dynamo_args.frontend_decoding and server_args.disaggregation_mode in (
+            None,
+            "null",
+        ):
+            publish_sglang_qwen_video_processor_contract(runtime_config, engine)
     # set reasoning parser and tool call parser
     runtime_config.reasoning_parser = dynamo_args.dyn_reasoning_parser
     runtime_config.tool_call_parser = dynamo_args.dyn_tool_call_parser

components/src/dynamo/sglang/request_handlers/llm/decode_handler.py

@@ -4,7 +4,7 @@
 import asyncio
 import logging
 import time
-from typing import Any, AsyncGenerator, AsyncIterator, Dict, List, Mapping, Optional
+from typing import Any, AsyncGenerator, AsyncIterator, Dict, Mapping, Optional
 
 import numpy as np
 import sglang as sgl
@@ -40,6 +40,7 @@
     IMAGE_URL_KEY,
     VIDEO_URL_KEY,
     build_disagg_mm_kwargs,
+    extract_mm_hashes,
     extract_media_urls,
     raise_if_unextracted_multimodal,
 )
@@ -305,32 +306,6 @@ def _resolve_mm_hashes_supported(engine: Any) -> bool:
         probe = filter_supported_async_generate_kwargs(engine, {"mm_hashes": None})
         return "mm_hashes" in probe
 
-    @staticmethod
-    def _extract_mm_hashes(request: Dict[str, Any]) -> Optional[List[str]]:
-        """Pull the per-image hashes the Rust frontend forwards via extra_args.
-
-        Returns ``None`` when the field is absent or malformed; SGLang then
-        recomputes the hash internally via ``hash_feature()``.
-        """
-        extra_args = request.get("extra_args")
-        if not isinstance(extra_args, dict):
-            return None
-        mm_hashes = extra_args.get("mm_hashes")
-        if not mm_hashes:
-            return None
-        if not isinstance(mm_hashes, list):
-            return None
-        # Fail closed if a non-string slipped into the list — downstream
-        # SGLang treats mm_hashes as List[str] and a bad element would
-        # crash the worker mid-request. Routing falls back to text-prefix.
-        if not all(isinstance(h, str) for h in mm_hashes):
-            logging.warning(
-                "extra_args.mm_hashes contained non-str entries; "
-                "ignoring routing-side hashes and letting SGLang recompute"
-            )
-            return None
-        return mm_hashes
-
     def _metadata_uploader_from_request(
         self, request: Dict[str, Any]
     ) -> MetadataUploader | None:
@@ -623,7 +598,7 @@ async def generate(
 
             mm_hashes_kwargs: Dict[str, Any] = {}
             if self._mm_hashes_supported:
-                forwarded = self._extract_mm_hashes(request)
+                forwarded = extract_mm_hashes(request)
                 if forwarded is not None:
                     mm_hashes_kwargs["mm_hashes"] = forwarded
 

components/src/dynamo/sglang/request_handlers/llm/mm_disagg_utils.py

@@ -20,6 +20,9 @@
 _SUPPORTED_MULTIMODAL_CONTENT_TYPES = frozenset(
     {IMAGE_URL_KEY, AUDIO_URL_KEY, VIDEO_URL_KEY}
 )
+# BaseMultiModalProcessorOutput.organize_results() builds SGLang's mm_items in
+# this order, independent of their order in the original prompt.
+_SGLANG_MM_ITEM_MODALITY_ORDER = ("image", "video", "audio")
 
 
 def _multi_modal_data(request: Dict[str, Any]) -> Dict[str, Any]:
@@ -124,6 +127,64 @@ def extract_media_urls(
     return urls or None
 
 
+def extract_mm_hashes(request: Dict[str, Any]) -> list[str] | None:
+    """Return frontend MM hashes in SGLang's modality-grouped item order."""
+    extra_args = request.get("extra_args")
+    if not isinstance(extra_args, dict):
+        return None
+
+    grouped = extra_args.get("mm_hashes_by_modality")
+    if grouped is not None:
+        if not isinstance(grouped, dict):
+            logger.warning(
+                "extra_args.mm_hashes_by_modality is not an object; "
+                "ignoring routing-side hashes and letting SGLang recompute"
+            )
+            return None
+
+        unknown_modalities = {
+            str(modality)
+            for modality, hashes in grouped.items()
+            if modality not in _SGLANG_MM_ITEM_MODALITY_ORDER and hashes
+        }
+        if unknown_modalities:
+            logger.warning(
+                "extra_args.mm_hashes_by_modality contains unsupported "
+                "modalities %s; ignoring routing-side hashes and letting "
+                "SGLang recompute",
+                sorted(unknown_modalities),
+            )
+            return None
+
+        flattened: list[str] = []
+        for modality in _SGLANG_MM_ITEM_MODALITY_ORDER:
+            hashes = grouped.get(modality)
+            if hashes is None:
+                continue
+            if not isinstance(hashes, list) or not all(
+                isinstance(value, str) for value in hashes
+            ):
+                logger.warning(
+                    "extra_args.mm_hashes_by_modality[%s] is not a string "
+                    "list; ignoring routing-side hashes and letting SGLang recompute",
+                    modality,
+                )
+                return None
+            flattened.extend(hashes)
+        return flattened or None
+
+    mm_hashes = extra_args.get("mm_hashes")
+    if not mm_hashes or not isinstance(mm_hashes, list):
+        return None
+    if not all(isinstance(value, str) for value in mm_hashes):
+        logger.warning(
+            "extra_args.mm_hashes contained non-str entries; ignoring "
+            "routing-side hashes and letting SGLang recompute"
+        )
+        return None
+    return mm_hashes
+
+
 def build_disagg_mm_kwargs(request: Dict[str, Any]) -> Dict[str, Any]:
     """Build media kwargs for a disaggregated worker's ``async_generate`` call.
 

components/src/dynamo/sglang/tests/test_sglang_frontend_decoding.py

@@ -505,6 +505,34 @@ async def fake_async_generate(**kwargs):
     assert captured["audio_data"] == ["https://example.com/a.wav"]
 
 
+@pytest.mark.asyncio
+async def test_aggregated_forwards_grouped_mm_hashes_in_sglang_item_order():
+    handler = _new_decode_handler(enable_frontend_decoding=False)
+    handler._mm_hashes_supported = True
+    captured: Dict[str, Any] = {}
+
+    async def fake_async_generate(**kwargs):
+        captured.update(kwargs)
+        return _empty_stream()
+
+    handler.engine = SimpleNamespace(async_generate=fake_async_generate)
+    request = {
+        "token_ids": [1, 2, 3],
+        "multi_modal_data": {},
+        "extra_args": {
+            "mm_hashes_by_modality": {
+                "video": ["video-a"],
+                "image": ["image-a"],
+            }
+        },
+    }
+
+    async for _ in handler.generate(request, _Context()):
+        pass
+
+    assert captured["mm_hashes"] == ["image-a", "video-a"]
+
+
 @pytest.mark.asyncio
 async def test_aggregated_fd_on_loads_decoded_variants_to_pil():
     """With --frontend-decoding, Decoded items are loaded via ImageLoader and

components/src/dynamo/sglang/tests/test_sglang_multimodal_utils.py

@@ -8,6 +8,7 @@
 from dynamo.llm.exceptions import InvalidArgument
 from dynamo.sglang.request_handlers.llm.mm_disagg_utils import (
     build_disagg_mm_kwargs,
+    extract_mm_hashes,
     extract_media_urls,
     raise_if_unextracted_multimodal,
 )
@@ -79,6 +80,41 @@ def test_extract_media_urls_rejects_malformed_payloads():
         extract_media_urls({"image_url": ""}, "image_url")
 
 
+def test_extract_mm_hashes_preserves_legacy_image_protocol():
+    request = {"extra_args": {"mm_hashes": ["image-a", "image-b"]}}
+
+    assert extract_mm_hashes(request) == ["image-a", "image-b"]
+
+
+def test_extract_mm_hashes_flattens_in_sglang_item_order():
+    request = {
+        "extra_args": {
+            "mm_hashes_by_modality": {
+                "video": ["video-a"],
+                "image": ["image-a", "image-b"],
+            }
+        }
+    }
+
+    assert extract_mm_hashes(request) == ["image-a", "image-b", "video-a"]
+
+
+@pytest.mark.parametrize(
+    "grouped",
+    [
+        ["not-an-object"],
+        {"video": "not-a-list"},
+        {"video": ["ok", 1]},
+        {"future_modality": ["hash"]},
+    ],
+)
+def test_extract_mm_hashes_rejects_malformed_grouped_protocol(grouped):
+    assert (
+        extract_mm_hashes({"extra_args": {"mm_hashes_by_modality": grouped}})
+        is None
+    )
+
+
 class TestMultimodalGuard:
     """Tests for multimodal guard when frontend extraction is missing."""
 

components/src/dynamo/sglang/tests/test_sglang_video_routing.py

@@ -0,0 +1,138 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from dynamo.llm import ModelRuntimeConfig
+from dynamo.sglang import video_routing
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.sglang,
+    pytest.mark.multimodal,
+    pytest.mark.gpu_0,
+    pytest.mark.profiled_vram_gib(0),
+    pytest.mark.pre_merge,
+]
+
+
+def _engine(
+    *,
+    video_config=None,
+    model_type="qwen3_vl",
+    architecture="Qwen3VLForConditionalGeneration",
+    overrides_video_replacement=True,
+):
+    if overrides_video_replacement:
+
+        class QwenProcessor(video_routing.transformers.ProcessorMixin):
+            def replace_video_token(self):
+                return None
+
+    else:
+
+        class QwenProcessor(video_routing.transformers.ProcessorMixin):
+            pass
+
+    mm_processor = SimpleNamespace(
+        hf_config=SimpleNamespace(
+            model_type=model_type,
+            architectures=[architecture],
+        ),
+        video_config=video_config or {},
+        _processor=QwenProcessor.__new__(QwenProcessor),
+    )
+    return SimpleNamespace(
+        tokenizer_manager=SimpleNamespace(mm_processor=mm_processor)
+    )
+
+
+@pytest.fixture
+def qwen_preprocessor(monkeypatch):
+    monkeypatch.setattr(
+        video_routing,
+        "sglang_qwen_vl",
+        SimpleNamespace(
+            IMAGE_FACTOR=28,
+            VIDEO_MIN_PIXELS=100352,
+            VIDEO_MAX_PIXELS=602112,
+            VIDEO_TOTAL_PIXELS=90316800,
+            FRAME_FACTOR=2,
+            FPS=2.0,
+            FPS_MIN_FRAMES=4,
+            FPS_MAX_FRAMES=768,
+        ),
+    )
+    monkeypatch.setattr(
+        video_routing,
+        "qwen3_smart_resize",
+        lambda **_: (1216, 4096),
+    )
+
+
+def test_publishes_sglang_qwen_video_contract(qwen_preprocessor):
+    runtime_config = ModelRuntimeConfig()
+
+    video_routing.publish_sglang_qwen_video_processor_contract(
+        runtime_config, _engine()
+    )
+
+    contract = json.loads(
+        runtime_config.runtime_data[
+            video_routing.SGLANG_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY
+        ]
+    )
+    assert contract == {
+        "placeholder_target": "bare_video_token",
+        "resize_mode": "legacy_ceil",
+        "runless_boundary_hash": "tokens_only",
+        "sglang_preprocess": {
+            "image_factor": 28,
+            "video_min_pixels": 100352,
+            "video_max_pixels": 602112,
+            "video_total_pixels": 90316800,
+            "frame_factor": 2,
+            "fps": 2.0,
+            "min_frames": 4,
+            "max_frames": 768,
+        },
+    }
+
+
+def test_processor_override_disables_exact_video_contract(qwen_preprocessor):
+    assert (
+        video_routing._resolve_qwen_video_processor_contract(
+            _engine(video_config={"fps": 1.0})
+        )
+        is None
+    )
+
+
+def test_inherited_video_replacement_publishes_wrapped_target(qwen_preprocessor):
+    contract = video_routing._resolve_qwen_video_processor_contract(
+        _engine(overrides_video_replacement=False)
+    )
+
+    assert contract is not None
+    assert contract["placeholder_target"] == "vision_wrapped_video_token"
+
+
+@pytest.mark.parametrize(
+    ("model_type", "architecture"),
+    [
+        ("llava", "LlavaForConditionalGeneration"),
+        ("qwen3_vl", "Qwen3VLForCausalLM"),
+    ],
+)
+def test_non_qwen_video_model_does_not_publish_contract(
+    qwen_preprocessor, model_type, architecture
+):
+    assert (
+        video_routing._resolve_qwen_video_processor_contract(
+            _engine(model_type=model_type, architecture=architecture)
+        )
+        is None
+    )