feat(sglang): enable KVCR linker inventory and hints

components/src/dynamo/sglang/kv_hints.py

@@ -15,6 +15,7 @@
     KV_HINT_TRANSFER_SOURCE_CONTROL_ENDPOINTS_RUNTIME_KEY,
     KV_HINT_TRANSFER_WORKER_TYPE_RUNTIME_KEY,
 )
+from dynamo.common.kv_state_agent import validate_advertise_host
 from dynamo.sglang._compat import (
     filter_supported_async_generate_kwargs,
     resolved_server_args,
@@ -30,7 +31,34 @@ def kv_hint_kwargs(engine: Any, request: Mapping[str, Any]) -> dict[str, Any]:
     return filter_supported_async_generate_kwargs(engine, {"kv_hints": dict(hint)})
 
 
-def kvcr_hint_endpoints(scheduler_infos, rank_bounds) -> dict[str, str]:
+def is_kvcr_linker(server_args) -> bool:
+    return (
+        getattr(server_args, "enable_unified_cache_external_linker", False)
+        and getattr(server_args, "unified_cache_external_linker_backend", None)
+        == "kvcr"
+        and not getattr(server_args, "enable_hierarchical_cache", False)
+    )
+
+
+def is_supported_kvcr_linker_topology(server_args) -> bool:
+    return is_kvcr_linker(server_args) and all(
+        getattr(server_args, name, 1) in (None, 1)
+        for name in (
+            "nnodes",
+            "tp_size",
+            "dp_size",
+            "attn_dp_size",
+            "attn_cp_size",
+            "dcp_size",
+            "pp_size",
+            "tokenizer_worker_num",
+        )
+    )
+
+
+def kvcr_hint_endpoints(
+    scheduler_infos, rank_bounds, *, require_inventory=False
+) -> dict[str, str]:
     """Validate SGLang-owned coordinates; never infer backend defaults or offsets."""
     endpoints: dict[str, str] = {}
     expected_ranks = set(range(*rank_bounds))
@@ -46,6 +74,13 @@ def kvcr_hint_endpoints(scheduler_infos, rank_bounds) -> dict[str, str]:
                 raise ValueError("SGLang KV hint capabilities must be a list")
             if "router_hint" not in capabilities:
                 continue
+            if require_inventory and (
+                "kvcr_inventory" not in capabilities
+                or source.get("inventory_lifetime") != "process_local"
+                or source.get("reset_clears_owner_inventory") is not True
+                or source.get("owner_incarnation_scoped") is not True
+            ):
+                continue
             rank = source.get("dp_rank")
             endpoint = source.get("source_control_endpoint")
             if type(rank) is not int or rank not in expected_ranks:
@@ -78,6 +113,9 @@ def kvcr_hint_endpoints(scheduler_infos, rank_bounds) -> dict[str, str]:
                     raise ValueError(
                         "SGLang KV hint source cannot advertise a wildcard"
                     )
+            if require_inventory:
+                host = parsed.hostname
+                validate_advertise_host(f"[{host}]" if ":" in host else host)
             key = str(rank)
             if key in endpoints and endpoints[key] != endpoint:
                 raise ValueError(
@@ -92,10 +130,11 @@ def kvcr_hint_endpoints(scheduler_infos, rank_bounds) -> dict[str, str]:
 def publish_kvcr_hint_capabilities(
     runtime_config, engine, server_args, worker_type
 ) -> None:
-    """Publish only HiCache sources explicitly reported by compatible SGLang."""
+    """Publish KVCR sources explicitly reported by a supported SGLang worker."""
     server_args = resolved_server_args(server_args)
+    linker = is_supported_kvcr_linker_topology(server_args)
     if (
-        not getattr(server_args, "enable_hierarchical_cache", False)
+        not (getattr(server_args, "enable_hierarchical_cache", False) or linker)
         or getattr(server_args, "nnodes", 1) != 1
         or getattr(server_args, "pp_size", 1) != 1
         or not filter_supported_async_generate_kwargs(engine, {"kv_hints": None})
@@ -105,7 +144,9 @@ def publish_kvcr_hint_capabilities(
     if init_result is None:
         return
     endpoints = kvcr_hint_endpoints(
-        init_result.scheduler_infos, model_card_dp_rank_bounds(server_args)
+        init_result.scheduler_infos,
+        model_card_dp_rank_bounds(server_args),
+        require_inventory=linker,
     )
     if not endpoints:
         return

components/src/dynamo/sglang/state_agent.py

@@ -1,6 +1,6 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
-"""Attach actual, process-local HiCache backend inventory to the KV state agent."""
+"""Attach actual, process-local KVCR inventory to the KV state agent."""
 
 from __future__ import annotations
 
@@ -14,11 +14,18 @@
     CACHE_OWNER_PATTERN,
     bounded_close,
     validate_advertise_host,
+    worker_raw_topic,
 )
 from dynamo.llm import KvStateAttachmentOwner
-from dynamo.sglang._compat import resolved_server_args
+from dynamo.sglang._compat import (
+    filter_supported_async_generate_kwargs,
+    resolved_server_args,
+)
 from dynamo.sglang.capacity import model_card_dp_rank_bounds
-from dynamo.sglang.kv_hints import kvcr_hint_endpoints
+from dynamo.sglang.kv_hints import (
+    is_supported_kvcr_linker_topology,
+    kvcr_hint_endpoints,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -55,17 +62,22 @@ def state_agent_descriptors(engine, config) -> list[dict]:
     if not sources:
         return []
     server_args = resolved_server_args(config.server_args)
+    linker = is_supported_kvcr_linker_topology(server_args)
     if (
-        not server_args.enable_hierarchical_cache
+        not (getattr(server_args, "enable_hierarchical_cache", False) or linker)
         or server_args.nnodes != 1
         or server_args.pp_size != 1
         or server_args.tokenizer_worker_num != 1
         or not config.dynamo_args.use_kv_events
     ):
         raise ValueError(
-            "KVCR inventory requires a single-host HiCache worker, one tokenizer "
+            "KVCR inventory requires a single-host HiCache or supported linker worker, one tokenizer "
             "worker, PP=1 and KV event emission"
         )
+    if linker and not filter_supported_async_generate_kwargs(
+        engine, {"kv_hints": None}
+    ):
+        raise ValueError("KVCR linker inventory requires a hint-forwarding worker")
     roles = {
         DisaggregationMode.AGGREGATED: "aggregated",
         DisaggregationMode.PREFILL: "prefill",
@@ -77,8 +89,19 @@ def state_agent_descriptors(engine, config) -> list[dict]:
     bounds = model_card_dp_rank_bounds(server_args)
     if set(sources) != set(range(*bounds)):
         raise ValueError("KVCR inventory must cover every managed DP rank")
+    if linker:
+        for source in sources.values():
+            if source.get("inventory_lifetime") != "process_local":
+                raise ValueError("KVCR linker must report process-local inventory")
+            if (
+                source.get("reset_clears_owner_inventory") is not True
+                or source.get("owner_incarnation_scoped") is not True
+            ):
+                raise ValueError(
+                    "KVCR linker inventory requires reset and incarnation fencing"
+                )
     infos = engine._scheduler_init_result.scheduler_infos
-    endpoints = kvcr_hint_endpoints(infos, bounds)
+    endpoints = kvcr_hint_endpoints(infos, bounds, require_inventory=linker)
     if set(endpoints) != {str(rank) for rank in sources}:
         raise ValueError(
             "KVCR inventory requires a router_hint source for every managed DP rank"
@@ -129,11 +152,15 @@ def state_agent_descriptors(engine, config) -> list[dict]:
         block_size = raw.get("block_size")
         if type(block_size) is not int or block_size <= 0:
             raise ValueError("KVCR raw event source requires a positive block size")
+        if linker and block_size != getattr(server_args, "page_size", None):
+            raise ValueError("KVCR linker event block size must match its page size")
         topic = raw.get("topic", "")
         if not isinstance(topic, str) or not topic.strip():
             raise ValueError(
                 "KVCR raw event topic must identify this engine incarnation"
             )
+        if linker and topic != worker_raw_topic([owner]):
+            raise ValueError("KVCR linker event topic must identify its owner")
         descriptors.append(
             {
                 "cache_owner_id": owner,
@@ -176,7 +203,7 @@ async def start(self, engine, config, endpoint):
             )
             raise
         self._owner = owner
-        logger.info("Attached %d KVCR HiCache inventory owners", len(descriptors))
+        logger.info("Attached %d KVCR inventory owners", len(descriptors))
 
     async def _close_after_start(self, owner, startup):
         # Let the native binding initialize its OnceCell before withdrawal.

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

@@ -15,7 +15,11 @@
     KV_HINT_TRANSFER_WORKER_TYPE_RUNTIME_KEY,
 )
 from dynamo.llm import ModelType, WorkerType
-from dynamo.sglang.kv_hints import kv_hint_kwargs, kvcr_hint_endpoints
+from dynamo.sglang.kv_hints import (
+    kv_hint_kwargs,
+    kvcr_hint_endpoints,
+    publish_kvcr_hint_capabilities,
+)
 
 # Marker-only collection may provide only partial SGLang stubs.
 _register_import_error = None
@@ -59,6 +63,59 @@ def test_router_hint_forwarding_is_private_and_lossless():
     assert kv_hint_kwargs(_HintEngine(), {"nvext": {"kv_hint": hint}}) == {}
 
 
+@pytest.mark.parametrize(
+    "worker_type", [WorkerType.Aggregated, WorkerType.Prefill, WorkerType.Decode]
+)
+def test_linker_hint_advertisement_requires_real_inventory_and_supported_engine(
+    worker_type,
+):
+    args = SimpleNamespace(
+        enable_hierarchical_cache=False,
+        enable_unified_cache_external_linker=True,
+        unified_cache_external_linker_backend="kvcr",
+        nnodes=1,
+        tp_size=1,
+        dp_size=1,
+        pp_size=1,
+    )
+    source = {
+        "dp_rank": 0,
+        "capabilities": ["router_hint", "kvcr_inventory"],
+        "source_control_endpoint": "tcp://worker:24100",
+        "inventory_lifetime": "process_local",
+        "reset_clears_owner_inventory": True,
+        "owner_incarnation_scoped": True,
+    }
+    engine = _HintEngine()
+    engine._scheduler_init_result = SimpleNamespace(
+        scheduler_infos=[{"kv_hint_sources": [source]}]
+    )
+    values = {}
+    runtime = SimpleNamespace(
+        set_engine_specific=lambda k, v: values.update({k: json.loads(v)})
+    )
+    publish_kvcr_hint_capabilities(runtime, engine, args, worker_type)
+    assert values == {
+        KV_HINT_TRANSFER_CAPABILITY_KEY: True,
+        KV_HINT_TRANSFER_WORKER_TYPE_RUNTIME_KEY: str(worker_type),
+        KV_HINT_TRANSFER_SOURCE_CONTROL_ENDPOINTS_RUNTIME_KEY: {
+            "0": "tcp://worker:24100"
+        },
+    }
+    values.clear()
+    source["capabilities"] = ["router_hint"]
+    publish_kvcr_hint_capabilities(runtime, engine, args, worker_type)
+    assert values == {}
+    source["capabilities"].append("kvcr_inventory")
+    args.tp_size = 2
+    publish_kvcr_hint_capabilities(runtime, engine, args, worker_type)
+    assert values == {}
+    args.tp_size = 1
+    args.dp_size = 2
+    publish_kvcr_hint_capabilities(runtime, engine, args, worker_type)
+    assert values == {}
+
+
 @pytest.mark.parametrize(
     "change",
     [