[None][feat] Add image routing metadata to OpenEngine KV events

cpp/tensorrt_llm/nanobind/batch_manager/bindings.cpp

@@ -238,6 +238,16 @@ void initBindings(nb::module_& m)
                 }
                 return hashes;
             })
+        .def_prop_ro("multimodal_uuids",
+            [](GenLlmReq& self)
+            {
+                std::optional<std::vector<std::optional<std::string>>> uuids = std::nullopt;
+                if (self.getMultimodalUuids())
+                {
+                    uuids = *self.getMultimodalUuids().value();
+                }
+                return uuids;
+            })
         .def_prop_ro("multimodal_positions",
             [](GenLlmReq& self)
             {

docs/source/features/kvcache.md

@@ -323,6 +323,16 @@ prompt = TextPrompt(
 
 V2 derives `mm_keys` directly from the cached token sequence. Each entry identifies a continuous multimodal segment within that block: `hash` is the item's digest, and `start_offset` is the segment's first token offset within the item. An item spanning multiple blocks retains the same digest with increasing offsets. Text may separate segments of the same item. Items are processed in prompt order; one item cannot resume after another item has started. The item digest is distinct from `block_hash`, which also depends on the preceding token sequence.
 
+Streaming V2 events can also include an optional `uuid` in each `mm_keys` entry.
+It is a bounded routing hint; it does not replace the content-sensitive digest.
+Missing, evicted, or conflicting hints omit `uuid`.
+For multimodal blocks, the optional `BlockStored.trtllm_image_routing` extension
+provides the routing format version and token-ID offset. The extension can accompany
+video and audio keys as well as image keys; the OpenEngine capability flag advertises
+only the currently validated image-routing support. See the
+[OpenEngine image routing contract](source:tensorrt_llm/grpc/openengine/README.md#image-routing-metadata)
+for capability requirements and consumer handling.
+
 
 ### Enable Offloading to Host Memory
 

tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py

@@ -6126,6 +6126,16 @@ def _augment_tokens_for_block_reuse(
                 "reuse."
             )
 
+        if chunk_start == 0 and isinstance(self.event_manager, StreamingKVCacheEventManager):
+            uuids = req.multimodal_uuids
+            if uuids is not None:
+                uuid_pairs = [
+                    (_hash_to_digest(digest), uuid)
+                    for digest, uuid in zip(req.multimodal_hashes, uuids, strict=True)
+                    if uuid is not None
+                ]
+                self.event_manager.remember_multimodal_uuids(uuid_pairs)
+
         # Multimodal path: materialize a Python-int list (digest bytes get spliced in below),
         # which flows through the per-element binding fallback. tokens may be a zero-copy numpy
         # int32 view (get_tokens_view) — use tolist() so elements are Python ints, not np.int32.

tensorrt_llm/_torch/pyexecutor/kv_cache_events.py

@@ -29,10 +29,10 @@
 import time
 import traceback
 from abc import ABC, abstractmethod
-from collections import deque
+from collections import OrderedDict, deque
 from itertools import count
 from queue import Queue
-from typing import Any, Iterator, Optional
+from typing import Any, Iterable, Iterator, Optional
 
 import msgspec
 import zmq
@@ -80,6 +80,14 @@ class MultimodalKey(
 
     hash: str
     start_offset: int
+    uuid: str | None = None
+
+
+class ImageRoutingMetadata(msgspec.Struct, frozen=True, gc=False):
+    """Versioned digest-first image normalization metadata for KV routers."""
+
+    version: int
+    token_id_offset: int
 
 
 class BlockStored(KVCacheWireEvent):
@@ -99,6 +107,7 @@ class BlockStored(KVCacheWireEvent):
     locality: str | None = None
     # Aligned one-for-one with block_hashes when multimodal decoding is enabled.
     mm_keys: list[list[MultimodalKey]] | None = None
+    trtllm_image_routing: ImageRoutingMetadata | None = None
 
 
 class BlockRemoved(KVCacheWireEvent):
@@ -551,6 +560,10 @@ def __init__(
             raise ValueError("mm_token_id_offset must be non-negative")
         self._block_size = block_size
         self._include_mm_keys = mm_token_id_offset is not None
+        self._mm_token_id_offset = mm_token_id_offset
+        self._uuid_capacity = max_entries
+        self._uuids: OrderedDict[bytes, str | None] = OrderedDict()
+        self._uuid_lock = threading.Lock()
         self._event_sink = kv_cache_manager_v2_runtime.StreamingEventSink(
             max_entries=max_entries,
             mm_token_id_offset=mm_token_id_offset,
@@ -560,13 +573,42 @@ def __init__(
     def event_sink(self) -> kv_cache_manager_v2_runtime.StreamingEventSink:
         return self._event_sink
 
+    def remember_multimodal_uuids(self, items: Iterable[tuple[bytes, str]]) -> None:
+        """Retain bounded routing hints without changing native cache keys."""
+        for digest, uuid in items:
+            if len(uuid.encode("utf-8")) > 256:
+                continue
+            with self._uuid_lock:
+                if digest in self._uuids and self._uuids[digest] != uuid:
+                    # Keep conflicting hints ambiguous until their bounded entry is evicted.
+                    self._uuids[digest] = None
+                else:
+                    self._uuids[digest] = uuid
+                self._uuids.move_to_end(digest)
+                while len(self._uuids) > self._uuid_capacity:
+                    self._uuids.popitem(last=False)
+
+    def _multimodal_key(self, key: tuple) -> MultimodalKey:
+        digest = bytes(key[0])
+        with self._uuid_lock:
+            uuid = self._uuids.get(digest)
+        return MultimodalKey(hash=digest.hex(), start_offset=key[1], uuid=uuid)
+
     def set_target_life_cycle(self, life_cycle_id: int) -> None:
         self._event_sink.set_target_life_cycle(life_cycle_id)
 
     def drain_events(self) -> list[BlockStored | BlockRemoved | AllBlocksCleared]:
         result: list[BlockStored | BlockRemoved | AllBlocksCleared] = []
         for event in self._event_sink.drain_iteration_events():
             if isinstance(event, kv_cache_manager_v2_runtime.StreamingBlockStoredData):
+                mm_keys = (
+                    [
+                        [self._multimodal_key(key) for key in block_keys]
+                        for block_keys in event.mm_keys
+                    ]
+                    if self._include_mm_keys
+                    else None
+                )
                 result.append(
                     BlockStored(
                         block_hashes=list(event.block_hashes),
@@ -576,15 +618,14 @@ def drain_events(self) -> list[BlockStored | BlockRemoved | AllBlocksCleared]:
                         lora_id=event.lora_id,
                         medium="GPU",
                         lora_name=None,
-                        mm_keys=(
-                            [
-                                [
-                                    MultimodalKey(hash=bytes(key[0]).hex(), start_offset=key[1])
-                                    for key in block_keys
-                                ]
-                                for block_keys in event.mm_keys
-                            ]
-                            if self._include_mm_keys
+                        mm_keys=mm_keys,
+                        trtllm_image_routing=(
+                            ImageRoutingMetadata(
+                                version=1, token_id_offset=self._mm_token_id_offset
+                            )
+                            if mm_keys is not None
+                            and self._mm_token_id_offset is not None
+                            and any(mm_keys)
                             else None
                         ),
                     )
@@ -646,6 +687,10 @@ def start(self) -> None:
         """Bind the publisher's sockets and start its background thread."""
         self._publisher.start()
 
+    def remember_multimodal_uuids(self, items: Iterable[tuple[bytes, str]]) -> None:
+        """Associate native MM digests with optional request identities."""
+        self._event_source.remember_multimodal_uuids(items)
+
     def get_source_descriptor(self) -> dict[str, str | int] | None:
         """Describe a live ZMQ publisher for OpenEngine node discovery."""
         publisher = self._publisher

tensorrt_llm/grpc/openengine/README.md

@@ -73,7 +73,7 @@ The `Inference.Generate` RPC loads the selected model through TensorRT-LLM's PyT
 
 Clients must continuously consume the response stream. If response delivery remains stalled for 30 seconds, the server aborts the engine request and terminates the stream with a retryable overload error.
 
-Features without a faithful TensorRT-LLM mapping return `UNIMPLEMENTED`: prefix-cache bypass, LoRA lifecycle selection, multimodal media, explicit-token or all-vocabulary log-probability selection, nonzero prompt-logprob offsets, per-request grammar-backend selection, and priority metadata. Attention-DP rank hints are validated and used for first-turn placement. With conversation affinity on, the first hint establishes the conversation's rank, and later turns retain that rank even if their hints differ; the request must carry a stable conversation ID. `ServerInfo.extra.trtllm_supports_dp_rank_targeting` advertises this support. The AutoDeploy backend is rejected at startup until it supports request cancellation. `Control` implements `GetServerInfo`, `GetModelInfo`, `GetLoad`, `GetKvEventSources`, `Health` and `Abort`; its LoRA lifecycle and `SubscribeKvEvents` RPCs return `UNIMPLEMENTED`.
+Features without a faithful TensorRT-LLM mapping return `UNIMPLEMENTED`: prefix-cache bypass, LoRA lifecycle selection, explicit-token or all-vocabulary log-probability selection, nonzero prompt-logprob offsets, per-request grammar-backend selection, and priority metadata. Attention-DP rank hints are validated and used for first-turn placement. With conversation affinity on, the first hint establishes the conversation's rank, and later turns retain that rank even if their hints differ; the request must carry a stable conversation ID. `ServerInfo.extra.trtllm_supports_dp_rank_targeting` advertises this support. The AutoDeploy backend is rejected at startup until it supports request cancellation. `Control` implements `GetServerInfo`, `GetModelInfo`, `GetLoad`, `GetKvEventSources`, `Health` and `Abort`; its LoRA lifecycle and `SubscribeKvEvents` RPCs return `UNIMPLEMENTED`.
 
 ### Disaggregated serving
 
@@ -118,6 +118,39 @@ Enable load reporting independently with `--grpc --grpc-protocol openengine --op
 
 For attention DP, each publisher binds `base_port + global_rank`. When event streaming is enabled, OpenEngine collects the live publisher descriptors at executor startup and advertises every rank, including remote nodes. Those hostnames and ports must be reachable from the subscriber. A missing publisher fails startup; an unavailable discovery snapshot fails the RPC rather than synthesizing socket addresses. PP and CP streaming remain unsupported. The extra placement collection is OpenEngine-only; native HTTP serving and OpenEngine with events disabled do not enable it.
 
+### Image routing metadata
+
+`GetModelInfo.extra.trtllm_image_routing_version = 1` advertises image routing
+metadata for aggregated Qwen3-VL with KV cache manager V2 explicitly enabled,
+block reuse enabled, and ZMQ event publishing. Image passthrough separately
+requires `supports_multimodal` and `extra.trtllm_supported_modalities` to advertise
+image support. Set `use_kv_cache_manager_v2=True` explicitly: `"auto"` is
+excluded from capability discovery even if the engine resolves it to V2.
+Video, audio, and disaggregated image routing are not advertised.
+
+For image blocks, `BlockStored.trtllm_image_routing` contains
+`version` and `token_id_offset` (the vocabulary size). The existing
+`BlockStored.mm_keys` contains the keys, aligned with the stored blocks. Each key
+retains its native content `hash` and `start_offset` and gains an optional `uuid`.
+Keys are serialized only in `mm_keys`. The publisher emits this extension for
+all multimodal keys, including video/audio keys; the name identifies the current
+image-routing contract, not a modality filter. Only image routing has been
+validated and advertised by OpenEngine. Native image tokens start with a digest
+and continue with IDs above `token_id_offset`; a router can normalize those
+tokens using identities it sent in `MediaItem.uuid`. Native block hashes, parent chains, and removals retain
+their existing meaning. Cache hashing continues to include image content.
+
+The publisher retains at most `max_entries` digest-to-UUID associations and skips
+UUIDs longer than 256 UTF-8 bytes. Conflicting hints mark a digest ambiguous until
+its entry is evicted or the publisher restarts, without failing inference.
+Every registration refreshes the entry's LRU position, including ambiguous entries,
+so a frequently registered digest can remain ambiguous for the publisher's lifetime.
+An agreeing registration does not restore a hint after a conflict. Missing, evicted,
+or ambiguous hints omit the UUID while retaining the image-routing metadata. Consumers
+must ignore unmapped image stores while continuing to process removals. Routing
+identities must refer to stable media content. This extension does not change the
+OpenEngine schema.
+
 ### Node-local discovery
 
 OpenEngine exposes optional, versioned `ServerInfo.extra.trtllm_node` metadata without changing the protocol schema. It describes a shared engine incarnation (`engine_id`), local hostname (`node_id`), whether this is the leader, node count, KV block size, local publishing DP ranks, and the global rank-to-node ownership map. The engine ID matches `GetServerInfo.instance_id` on the serving ingress and node metadata listeners. Consumers can use this information to partition telemetry collection without introducing routing-framework-specific server options.

tensorrt_llm/grpc/openengine/capabilities.py

@@ -67,6 +67,22 @@ def _loaded_model_type(llm: Any) -> str:
     return resolve_top_level_model_type(llm.input_processor.config)
 
 
+def supports_image_routing(llm: Any, modalities: Collection[str]) -> bool:
+    """Whether the loaded model exposes the validated image KV-event contract."""
+    processor_config = getattr(getattr(llm, "input_processor", None), "config", None)
+    if "image" not in modalities or resolve_top_level_model_type(processor_config) != "qwen3_vl":
+        return False
+    cache = getattr(llm.args, "kv_cache_config", None)
+    events = getattr(cache, "kv_events_config", None)
+    return (
+        getattr(llm.args, "cache_transceiver_config", None) is None
+        and getattr(cache, "use_kv_cache_manager_v2", False) is True
+        and getattr(cache, "enable_block_reuse", False) is True
+        and getattr(events, "enable_kv_cache_events", False) is True
+        and getattr(events, "publisher", None) == "zmq"
+    )
+
+
 def resolve_media_capabilities(
     llm: Any, implemented_modalities: Collection[str]
 ) -> MediaCapabilities:

tensorrt_llm/grpc/openengine/control.py

@@ -35,7 +35,7 @@
     openengine_pb2_grpc,
     server_pb2,
 )
-from .capabilities import supported_guides
+from .capabilities import supported_guides, supports_image_routing
 from .coordinator import CoordinationError, FrontendClient, InvalidRequestIdError
 from .errors import AbortFailedError
 from .kv_events import data_parallel_size, events_config
@@ -296,6 +296,8 @@ async def GetModelInfo(
             reasoning_parser=str(self._llm.args.reasoning_parser or ""),
         )
         info.extra.update({"trtllm_supported_modalities": sorted(media_modalities)})
+        if supports_image_routing(self._llm, media_modalities):
+            info.extra.update({"trtllm_image_routing_version": 1})
 
         max_seq_len = _positive_int(self._llm.args.max_seq_len)
         if max_seq_len is not None:

tests/integration/test_lists/test-db/l0_a10.yml

@@ -69,6 +69,7 @@ l0_a10:
   # false advertisement by construction. The image case uses Qwen3-VL-2B.
   - unittest/grpc/openengine/test_capability_conformance.py
   - unittest/grpc/openengine/test_multimodal_conformance.py
+  - unittest/grpc/openengine/test_image_routing_conformance.py
   - unittest/disaggregated/test_coordinator_e2e.py
   - unittest/disaggregated/test_coordinator_worker.py
   - unittest/disaggregated/test_agent_multi_backends.py

tests/integration/test_lists/test-db/l0_cpu.yml

@@ -211,6 +211,7 @@ l0_cpu:
   - unittest/_torch/test_shared_tensor_cpu.py
   - unittest/grpc/openengine/test_frontend_lifecycle.py
   - unittest/grpc/openengine/test_control.py
+  - unittest/grpc/openengine/test_image_routing.py
   - unittest/grpc/openengine/test_multimodal.py
   - unittest/grpc/openengine/test_node.py
   - unittest/grpc/openengine/test_openengine.py

tests/unittest/_torch/executor/kv_cache/test_kv_cache_v2_first_new_block_probe.py

@@ -52,6 +52,7 @@ def make_stub_manager(
     # for a one-model draft, which reads D prompt tokens past the target's end.
     mgr.reuse_match_backoff = reuse_match_backoff
     mgr.vocab_size = 32000
+    mgr.event_manager = None
     mgr.conversation_manager = None
     mgr.kv_connector_manager = None
     mgr.kv_cache_map = {}

tests/unittest/_torch/executor/kv_cache/test_kv_cache_v2_multimodal_runs.py

@@ -23,6 +23,7 @@
 def _make_manager(vocab_size: int):
     manager = KVCacheManagerV2.__new__(KVCacheManagerV2)
     manager.vocab_size = vocab_size
+    manager.event_manager = None
     return manager