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

components/src/dynamo/common/kv_state_agent.py

@@ -3,14 +3,77 @@
 
 """Attachment helpers shared by KV state-agent workers."""
 
+import asyncio
 import hashlib
-from collections.abc import Iterable
+import ipaddress
+import logging
+import re
+import socket
+from collections.abc import Awaitable, Iterable
+
+logger = logging.getLogger(__name__)
+
+CACHE_OWNER_PATTERN = re.compile(
+    r"^(?P<domain>[0-9a-f]{32}:[0-9a-f]{32})/[0-9a-f]{16}/[0-9a-f]{32}$"
+)
+_STATE_AGENT_CLOSE_TIMEOUT_SECS = 30.0
 
 # Fixed width, so no worker's topic is a prefix of another's: SUB subscriptions
 # are prefix filters, and the state agent routes pooled raw streams by topic.
 _RAW_TOPIC_HEX_DIGITS = 16
 
 
+async def bounded_close(cleanup: Awaitable[None]) -> None:
+    """Bound cleanup without replacing a worker's error or cancellation on timeout."""
+    try:
+        await asyncio.wait_for(cleanup, timeout=_STATE_AGENT_CLOSE_TIMEOUT_SECS)
+    except asyncio.TimeoutError:
+        logger.warning(
+            "KV state attachment cleanup timed out after %.0fs; continuing worker shutdown",
+            _STATE_AGENT_CLOSE_TIMEOUT_SECS,
+        )
+
+
+def validate_advertise_host(host: str) -> None:
+    """Require a connectable hostname, IPv4 address, or bracketed IPv6 literal."""
+    if host.rstrip(".").lower() in {"*", "localhost", "0.0.0.0", "::", "[::]"}:
+        raise ValueError("raw_advertise_host must be a non-loopback connectable host")
+    if "://" in host or "/" in host:
+        raise ValueError("raw_advertise_host must not include a scheme, port, or path")
+    if host.startswith("["):
+        if not host.endswith("]"):
+            raise ValueError("raw_advertise_host has mismatched IPv6 brackets")
+        literal = host[1:-1]
+        try:
+            address = ipaddress.ip_address(literal)
+        except ValueError as error:
+            raise ValueError(
+                "bracketed raw_advertise_host must be an IPv6 literal"
+            ) from error
+        if address.version != 6:
+            raise ValueError("only IPv6 literals may use brackets")
+    else:
+        if "[" in host or "]" in host or ":" in host or any(c.isspace() for c in host):
+            raise ValueError(
+                "raw_advertise_host must be a hostname or bracketed IPv6 literal without a port"
+            )
+        literal = host
+        try:
+            address = ipaddress.ip_address(literal)
+        except ValueError:
+            # POSIX accepts abbreviated/integer IPv4 literals (e.g.127.1).
+            # Check them without resolving arbitrary DNS names at startup.
+            try:
+                address = ipaddress.IPv4Address(socket.inet_aton(literal))
+            except OSError:
+                return
+
+    if address.version == 6 and address.ipv4_mapped is not None:
+        address = address.ipv4_mapped
+    if address.is_loopback or address.is_unspecified:
+        raise ValueError("raw_advertise_host must be a non-loopback connectable host")
+
+
 def worker_raw_topic(cache_owner_ids: Iterable[str]) -> str:
     """Raw KV event topic of one worker, published by every DP rank it owns.
 

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()