[None][feat] support C++ streaming KV events with multimodal payloads

cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/CMakeLists.txt

@@ -56,7 +56,9 @@ set(KV_CACHE_MANAGER_V2_SRCS
     kv_cache_manager_v2/blockRadixTree.cpp
     kv_cache_manager_v2/batchedPageCopy.cu
     kv_cache_manager_v2/coldPageCodec.cpp
+    kv_cache_manager_v2/eventData.cpp
     kv_cache_manager_v2/eventManager.cpp
+    kv_cache_manager_v2/streamingEventSink.cpp
     kv_cache_manager_v2/page.cpp
     kv_cache_manager_v2/storageManager.cpp
     kv_cache_manager_v2/introspection.cpp

cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/eventData.cpp

@@ -0,0 +1,90 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "kv_cache_manager_v2/eventData.h"
+
+#include "kv_cache_manager_v2/blockRadixTree.h"
+
+#include <cstddef>
+#include <cstdint>
+
+namespace tensorrt_llm::batch_manager::kv_cache_manager_v2
+{
+
+std::string digestToHex(Digest const& digest)
+{
+    constexpr char kHex[] = "0123456789abcdef";
+    std::string result(digest.size() * 2, '\0');
+    auto out = result.begin();
+    for (auto byte : digest)
+    {
+        auto const value = std::to_integer<unsigned>(byte);
+        *out++ = kHex[value >> 4U];
+        *out++ = kHex[value & 0x0FU];
+    }
+    return result;
+}
+
+DecodedEventBlock decodeEventBlock(Block const& block, std::optional<int> mmTokenIdOffset)
+{
+    DecodedEventBlock result;
+    result.tokenIds.reserve(block.tokens.size());
+
+    Digest const* itemDigest = nullptr;
+    if (mmTokenIdOffset.has_value() && block.prev != nullptr && block.prev->type() == NodeBase::Type::kBLOCK)
+    {
+        itemDigest = static_cast<Block const*>(block.prev)->getLastTokenDigest().get();
+    }
+    bool inMmRun = false;
+    for (auto const& token : block.tokens)
+    {
+        if (token.isDigest())
+        {
+            result.tokenIds.emplace_back(std::in_place_index<1>, digestToHex(token.digest()));
+            if (mmTokenIdOffset.has_value())
+            {
+                itemDigest = &token.digest();
+                result.mmKeys.push_back(
+                    {std::string(reinterpret_cast<char const*>(itemDigest->data()), itemDigest->size()), 0,
+                        std::nullopt, false});
+                inMmRun = true;
+            }
+            continue;
+        }
+
+        auto const tokenId = token.tokenId();
+        result.tokenIds.emplace_back(std::in_place_index<0>, tokenId);
+        if (itemDigest != nullptr && tokenId > *mmTokenIdOffset)
+        {
+            if (!inMmRun)
+            {
+                result.mmKeys.push_back(
+                    {std::string(reinterpret_cast<char const*>(itemDigest->data()), itemDigest->size()),
+                        tokenId - *mmTokenIdOffset, std::nullopt, false});
+            }
+            inMmRun = true;
+        }
+        else
+        {
+            // Text separates runs of the same item, so retain its digest for later continuations.
+            inMmRun = false;
+        }
+    }
+    return result;
+}
+
+} // namespace tensorrt_llm::batch_manager::kv_cache_manager_v2

cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/eventData.h

@@ -0,0 +1,61 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "kv_cache_manager_v2/common.h"
+
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <variant>
+#include <vector>
+
+namespace tensorrt_llm::batch_manager::kv_cache_manager_v2
+{
+
+struct Block;
+
+using EventTokenId = std::variant<int64_t, std::string>;
+
+struct MmKey
+{
+    std::string hash;
+    int startOffset = 0;
+    std::optional<std::string> uuid;
+    bool hasUuidField = false;
+
+    bool operator==(MmKey const& other) const
+    {
+        return hash == other.hash && startOffset == other.startOffset && uuid == other.uuid
+            && hasUuidField == other.hasUuidField;
+    }
+};
+
+struct DecodedEventBlock
+{
+    std::vector<EventTokenId> tokenIds;
+    std::vector<MmKey> mmKeys;
+};
+
+[[nodiscard]] std::string digestToHex(Digest const& digest);
+
+//! Decode the V2 digest-first multimodal representation used by KV event consumers.
+//! When mmTokenIdOffset is absent, tokens are still preserved but no MM segments are derived.
+[[nodiscard]] DecodedEventBlock decodeEventBlock(Block const& block, std::optional<int> mmTokenIdOffset);
+
+} // namespace tensorrt_llm::batch_manager::kv_cache_manager_v2

cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/eventManager.cpp

@@ -485,20 +485,6 @@ int EventManager::getWindowSize(EventLayerGroupId layerGroupId) const
     return windowSize == mWindowSizeByLayerGroup.end() ? mWindowSize : windowSize->second;
 }
 
-std::string EventManager::digestToHex(Digest const& digest)
-{
-    constexpr char kHex[] = "0123456789abcdef";
-    std::string result;
-    result.resize(digest.size() * 2);
-    for (size_t i = 0; i < digest.size(); ++i)
-    {
-        auto const value = std::to_integer<uint8_t>(digest[i]);
-        result[2 * i] = kHex[value >> 4U];
-        result[2 * i + 1] = kHex[value & 0x0FU];
-    }
-    return result;
-}
-
 uint64_t EventManager::truncateDigestToInt64(Digest const& digest)
 {
     uint64_t result = 0;
@@ -566,54 +552,15 @@ std::optional<KVCacheStoredBlockData> EventManager::storedBlockFromBlock(
         return std::nullopt;
     }
 
-    std::vector<MmKey> mmKeys;
-    Digest const* itemDigest = nullptr;
-    if (mMmTokenIdOffset.has_value() && block.prev != nullptr && block.prev->type() == NodeBase::Type::kBLOCK)
-    {
-        itemDigest = static_cast<Block const*>(block.prev)->getLastTokenDigest().get();
-    }
-    bool inMmRun = false;
+    auto decoded = decodeEventBlock(block, mMmTokenIdOffset);
     std::vector<UniqueToken> tokens;
-    tokens.reserve(block.tokens.size());
-    for (auto const& token : block.tokens)
+    tokens.reserve(decoded.tokenIds.size());
+    for (auto& tokenId : decoded.tokenIds)
     {
-        if (!token.isDigest())
-        {
-            UniqueToken uniqueToken;
-            uniqueToken.tokenId = EventTokenId{std::in_place_index<0>, token.tokenId()};
-            tokens.push_back(std::move(uniqueToken));
-            if (itemDigest != nullptr && token.tokenId() > *mMmTokenIdOffset)
-            {
-                if (!inMmRun)
-                {
-                    mmKeys.push_back(
-                        {std::string(reinterpret_cast<char const*>(itemDigest->data()), itemDigest->size()),
-                            token.tokenId() - *mMmTokenIdOffset, std::nullopt, false});
-                }
-                inMmRun = true;
-            }
-            else
-            {
-                // Text separates runs of the same item, so retain its digest for later continuations.
-                inMmRun = false;
-            }
-        }
-        else
-        {
-            UniqueToken uniqueToken;
-            uniqueToken.tokenId = EventTokenId{std::in_place_index<1>, digestToHex(token.digest())};
-            tokens.push_back(std::move(uniqueToken));
-            if (mMmTokenIdOffset.has_value())
-            {
-                itemDigest = &token.digest();
-                mmKeys.push_back({std::string(reinterpret_cast<char const*>(itemDigest->data()), itemDigest->size()), 0,
-                    std::nullopt, false});
-                inMmRun = true;
-            }
-        }
+        tokens.push_back(UniqueToken{std::move(tokenId)});
     }
     return KVCacheStoredBlockData{
-        hashFromBlock(block), std::move(tokens), cacheLevel.value(), priority, std::move(mmKeys), std::nullopt};
+        hashFromBlock(block), std::move(tokens), cacheLevel.value(), priority, std::move(decoded.mmKeys), std::nullopt};
 }
 
 uint64_t EventManager::hashV1BlockKey(std::vector<TokenId> const& tokens, uint64_t parentHash,

cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/eventManager.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include "kv_cache_manager_v2/common.h"
+#include "kv_cache_manager_v2/eventData.h"
 #include "kv_cache_manager_v2/eventSink.h"
 
 #include <condition_variable>
@@ -40,7 +41,6 @@ namespace tensorrt_llm::batch_manager::kv_cache_manager_v2
 {
 
 using EventBlockHash = std::variant<uint64_t, std::string>;
-using EventTokenId = std::variant<int64_t, std::string>;
 using EventLayerGroupId = std::optional<int>;
 
 struct UniqueToken
@@ -64,20 +64,6 @@ struct KVCacheCreatedData
     }
 };
 
-struct MmKey
-{
-    std::string hash;
-    int startOffset = 0;
-    std::optional<std::string> uuid;
-    bool hasUuidField = false;
-
-    bool operator==(MmKey const& other) const
-    {
-        return hash == other.hash && startOffset == other.startOffset && uuid == other.uuid
-            && hasUuidField == other.hasUuidField;
-    }
-};
-
 struct KVCacheStoredBlockData
 {
     EventBlockHash blockHash;
@@ -220,7 +206,6 @@ class EventManager final : public EventSink
     using V1RootAttrs = std::pair<std::optional<LoraTaskIdType>, std::optional<std::uint64_t>>;
 
     static std::pair<HashAlgorithm, std::string> parseHashAlgorithm(std::string const& hashAlgo);
-    static std::string digestToHex(Digest const& digest);
     static uint64_t truncateDigestToInt64(Digest const& digest);
     static std::vector<KVCacheEvent> trimEvents(std::vector<KVCacheEvent> events, int maxKvEventEntries);