feat(sglang): route KVCR HiCache inventory through state agents

components/src/dynamo/sglang/README.md

@@ -4,3 +4,36 @@
 # SGLang
 
 See [docs/backends/sglang/](../../../../docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/sglang/overview.md) for documentation.
+
+## KVCR-owned HiCache inventory
+
+KVCR inventory publication is opt-in for the in-process `python -m dynamo.sglang`
+HiCacheStorage path. It requires a matching SGLang backend that reports the
+`kvcr_inventory` capability and engine-lifetime cache-owner metadata.
+
+Keep KV events enabled and configure the KVCR backend with
+`"publish_inventory": true`, remote hints enabled, and an explicit reachable
+`control_advertise_host`. Use a ZMQ KV-event publisher with `"format": "dynamo"`.
+Start a state-agent host in the workers' namespace before starting the workers:
+
+```bash
+DYN_NAMESPACE=<worker-namespace> python -m dynamo.kv_state_agent
+```
+
+The worker reads the raw-event endpoint, incarnation-specific topic, cache owner,
+and peer-control endpoint from SGLang startup metadata. It attaches before model
+registration and uses `state_agent_v2` instead of the legacy event publisher.
+The router can then distinguish framework GPU/HiCache-host residency from actual
+KVCR-owned DRAM residency, even though both host caches use CPU memory.
+
+Upgrade the Dynamo frontend, worker, and state-agent host together when enabling
+this mode. Existing deployments without the capability keep the legacy path;
+vLLM attachment descriptors without an ingress protocol keep their existing
+default. The new attachment requires a single-host LLM engine, PP=1, and one
+tokenizer worker; incomplete source metadata is rejected at startup.
+
+This is a live process-local inventory, not durable recovery: detaching the
+engine clears its KVCR owner, and a detected raw-event sequence gap withdraws
+readiness until the worker is restarted. Router hints remain advisory; storage
+lookup and completed delivery still determine usable pages. This integration
+does not enable direct-device linker mode.

components/src/dynamo/sglang/init_llm.py

@@ -34,6 +34,7 @@
 )
 from dynamo.sglang.register import register_model_with_readiness_gate
 from dynamo.sglang.request_handlers import DecodeWorkerHandler, PrefillWorkerHandler
+from dynamo.sglang.state_agent import StateAgentLifecycle
 
 
 async def _warmup_prefill_engine(
@@ -201,7 +202,9 @@ async def init_decode(
         )
 
     body_failed = True
+    state_agent = StateAgentLifecycle()
     try:
+        await state_agent.start(engine, config, generate_endpoint)
         gather_tasks = [
             generate_endpoint.serve_endpoint(
                 handler.generate,
@@ -246,12 +249,15 @@ async def init_decode(
     else:
         body_failed = False
     finally:
-        await finish_worker_teardown(
-            metrics_task,
-            handler.cleanup,
-            run_deferred_handlers,
-            body_failed=body_failed,
-        )
+        try:
+            await state_agent.close()
+        finally:
+            await finish_worker_teardown(
+                metrics_task,
+                handler.cleanup,
+                run_deferred_handlers,
+                body_failed=body_failed,
+            )
 
 
 async def init_prefill(
@@ -354,7 +360,9 @@ async def init_prefill(
     ready_event = asyncio.Event()
 
     body_failed = True
+    state_agent = StateAgentLifecycle()
     try:
+        await state_agent.start(engine, config, generate_endpoint)
         await asyncio.gather(
             generate_endpoint.serve_endpoint(
                 handler.generate,
@@ -406,9 +414,12 @@ async def init_prefill(
     else:
         body_failed = False
     finally:
-        await finish_worker_teardown(
-            metrics_task,
-            handler.cleanup,
-            run_deferred_handlers,
-            body_failed=body_failed,
-        )
+        try:
+            await state_agent.close()
+        finally:
+            await finish_worker_teardown(
+                metrics_task,
+                handler.cleanup,
+                run_deferred_handlers,
+                body_failed=body_failed,
+            )

components/src/dynamo/sglang/publisher.py

@@ -49,6 +49,7 @@
     owns_engine_metrics,
 )
 from dynamo.sglang.nixl_telemetry import scheduler_metrics_ipc_name
+from dynamo.sglang.state_agent import inventory_sources
 
 
 def get_local_dp_rank_range(server_args) -> range:
@@ -396,6 +397,10 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]:
             List of KvEventPublisher instances if KV event publishing is enabled,
             empty list otherwise.
         """
+        if inventory_sources(self.engine):
+            # The versioned state agent consumes the mixed framework/KVCR
+            # stream. The legacy normalizer rejects independently owned events.
+            return []
         if self.dynamo_args.use_kv_events and not publishes_kv_events(self.server_args):
             logging.info(
                 "Non-leader node (node_rank=%s) shares the leader's single KV "

components/src/dynamo/sglang/register.py

@@ -64,6 +64,7 @@
     gateway_engine_id,
 )
 from dynamo.sglang.kv_hints import publish_kvcr_hint_capabilities
+from dynamo.sglang.state_agent import inventory_sources
 from dynamo.sglang.video_routing import publish_sglang_qwen_video_processor_contract
 
 SGLANG_HICACHE_MOONCAKE_RUNTIME_KEY = "sglang_hicache_mooncake"
@@ -212,6 +213,8 @@ async def _register_model_with_runtime_config(
     runtime_config = await get_runtime_config(engine, server_args, dynamo_args)
     if runtime_config is not None and forwards_kv_hints:
         publish_kvcr_hint_capabilities(runtime_config, engine, server_args, worker_type)
+        if inventory_sources(engine):
+            runtime_config.kv_event_source_mode = "state_agent_v2"
 
     if dynamo_args.use_sglang_tokenizer and not (
         output_type.supports_embedding() or output_type.supports_rerank()

components/src/dynamo/sglang/state_agent.py

@@ -0,0 +1,194 @@
+# 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."""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import logging
+import re
+from collections.abc import Mapping
+from urllib.parse import urlsplit
+
+from dynamo.common.constants import DisaggregationMode
+from dynamo.llm import KvStateAttachmentOwner
+from dynamo.sglang._compat import resolved_server_args
+from dynamo.sglang.capacity import model_card_dp_rank_bounds
+from dynamo.sglang.kv_hints import kvcr_hint_endpoints
+
+logger = logging.getLogger(__name__)
+_OWNER_PATTERN = re.compile(r"^[0-9a-f]{32}:[0-9a-f]{32}/[0-9a-f]{16}/[0-9a-f]{32}$")
+
+
+def inventory_sources(engine) -> dict[int, Mapping[str, object]]:
+    """Read capabilities from scheduler metadata, never backend configuration."""
+    initialization = getattr(engine, "_scheduler_init_result", None)
+    if initialization is None:
+        return {}
+    sources: dict[int, Mapping[str, object]] = {}
+    for info in initialization.scheduler_infos:
+        hints = info.get("kv_hint_sources", [])
+        if not isinstance(hints, list):
+            raise TypeError("SGLang KV hint sources must be a list")
+        for source in hints:
+            if not isinstance(source, Mapping):
+                raise TypeError("SGLang KV hint source must be an object")
+            capabilities = source.get("capabilities", [])
+            if not isinstance(capabilities, list):
+                raise TypeError("SGLang KV hint capabilities must be a list")
+            if "kvcr_inventory" not in capabilities:
+                continue
+            rank = source.get("dp_rank")
+            if type(rank) is not int or rank < 0:
+                raise ValueError("KVCR inventory source requires a nonnegative DP rank")
+            if rank in sources and sources[rank] != source:
+                raise ValueError("Conflicting KVCR inventory sources for one DP rank")
+            sources[rank] = source
+    return sources
+
+
+def state_agent_descriptors(engine, config) -> list[dict]:
+    sources = inventory_sources(engine)
+    if not sources:
+        return []
+    server_args = resolved_server_args(config.server_args)
+    if (
+        not server_args.enable_hierarchical_cache
+        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 "
+            "worker, PP=1 and KV event emission"
+        )
+    roles = {
+        DisaggregationMode.AGGREGATED: "aggregated",
+        DisaggregationMode.PREFILL: "prefill",
+        DisaggregationMode.DECODE: "decode",
+    }
+    role = roles.get(config.serving_mode)
+    if role is None:
+        raise ValueError("KVCR inventory requires a hint-forwarding LLM worker")
+    bounds = model_card_dp_rank_bounds(server_args)
+    if set(sources) != set(range(*bounds)):
+        raise ValueError("KVCR inventory must cover every managed DP rank")
+    infos = engine._scheduler_init_result.scheduler_infos
+    endpoints = kvcr_hint_endpoints(infos, bounds)
+    if set(endpoints) != {str(rank) for rank in sources}:
+        raise ValueError(
+            "KVCR inventory requires a router_hint source for every managed DP rank"
+        )
+    raw_sources: dict[int, Mapping[str, object]] = {}
+    for info in infos:
+        for source in info.get("kv_event_sources", []):
+            rank = source.get("dp_rank")
+            if type(rank) is not int or rank not in sources:
+                raise ValueError("KVCR raw event source has an invalid DP rank")
+            if rank in raw_sources and raw_sources[rank] != source:
+                raise ValueError("Conflicting raw KV event sources for one DP rank")
+            raw_sources[rank] = source
+    if set(raw_sources) != set(sources):
+        raise ValueError("KVCR inventory requires a reported raw event source per rank")
+    descriptors = []
+    owners = set()
+    for rank, source in sorted(sources.items()):
+        owner = source.get("cache_owner_id")
+        if not isinstance(owner, str) or _OWNER_PATTERN.fullmatch(owner) is None:
+            raise ValueError("KVCR inventory source requires a canonical CacheOwnerId")
+        if owner in owners:
+            raise ValueError("KVCR inventory owners must be unique across DP ranks")
+        owners.add(owner)
+        raw = raw_sources[rank]
+        if raw.get("format") != "dynamo":
+            raise ValueError("KVCR inventory requires Dynamo-format KV events")
+        endpoint = raw.get("endpoint")
+        if not isinstance(endpoint, str):
+            raise TypeError("KVCR inventory requires a reported TCP event endpoint")
+        parsed = urlsplit(endpoint)
+        if (
+            parsed.scheme != "tcp"
+            or not parsed.hostname
+            or parsed.port is None
+            or not 0 < parsed.port <= 65535
+            or parsed.username is not None
+            or parsed.password is not None
+            or parsed.path
+            or parsed.query
+            or parsed.fragment
+            or any(char.isspace() for char in parsed.hostname)
+            or parsed.hostname == "*"
+        ):
+            raise ValueError("Invalid reported KVCR raw event endpoint")
+        try:
+            address = ipaddress.ip_address(parsed.hostname)
+        except ValueError:
+            address = None
+        if address is not None and address.is_unspecified:
+            raise ValueError("KVCR raw event source cannot advertise a wildcard")
+        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")
+        topic = raw.get("topic", "")
+        if not isinstance(topic, str) or not topic.strip():
+            raise ValueError(
+                "KVCR raw event topic must identify this engine incarnation"
+            )
+        descriptors.append(
+            {
+                "cache_owner_id": owner,
+                "global_dp_rank": rank,
+                "kv_state_endpoint": config.dynamo_args.kv_state_endpoint
+                or f"{config.dynamo_args.namespace}/{config.dynamo_args.component}/"
+                f"{config.dynamo_args.endpoint}",
+                "indexer_domain_id": owner.split("/", 1)[0],
+                "kv_block_size": block_size,
+                "ingress_protocol": "sglang_hicache_v1",
+                "raw_zmq_endpoint": endpoint,
+                "raw_topic": topic,
+                "image_token_id": None,
+                "video_token_id": None,
+                "router_hint_source": {
+                    "source_control_endpoint": endpoints[str(rank)],
+                    "worker_type": role,
+                },
+            }
+        )
+    return descriptors
+
+
+class StateAgentLifecycle:
+    def __init__(self):
+        self._owner = None
+        self._cleanup_task = None
+
+    async def start(self, engine, config, endpoint):
+        descriptors = state_agent_descriptors(engine, config)
+        if not descriptors:
+            return
+        owner = KvStateAttachmentOwner(endpoint, endpoint.connection_id(), descriptors)
+        startup = asyncio.ensure_future(owner.start())
+        try:
+            await asyncio.shield(startup)
+        except asyncio.CancelledError:
+            self._cleanup_task = asyncio.create_task(
+                self._close_after_start(owner, startup)
+            )
+            raise
+        self._owner = owner
+        logger.info("Attached %d KVCR HiCache inventory owners", len(descriptors))
+
+    async def _close_after_start(self, owner, startup):
+        # Let the native binding initialize its OnceCell before withdrawal.
+        await startup
+        await owner.close()
+
+    async def close(self):
+        if self._cleanup_task is not None:
+            await asyncio.wait_for(asyncio.shield(self._cleanup_task), timeout=30.0)
+            return
+        owner, self._owner = self._owner, None
+        if owner is not None:
+            await asyncio.wait_for(owner.close(), timeout=30.0)