feat(kv-router): add Mooncake Store shared-cache routing

.github/codeowners/areas.yaml

@@ -1064,6 +1064,10 @@ shared:
   owners:
   - router
   - kv-memory
+- glob: lib/llm/src/kv_router/shared_cache/
+  owners:
+  - router
+  - kv-memory
 - glob: lib/llm/src/preprocessor/mm_routing/
   owners:
   - frontend

CODEOWNERS

@@ -644,6 +644,7 @@ requirements.txt
 /lib/kvbm-engine/src/pubsub/nats.rs                                                                     @ai-dynamo/dynamo-kv-memory-codeowners @ai-dynamo/dynamo-runtime-codeowners
 /lib/backend-common/src/telemetry.rs                                                                    @ai-dynamo/dynamo-runtime-codeowners @ai-dynamo/dynamo-observability-codeowners
 /lib/llm/src/http/service/metrics.rs                                                                    @ai-dynamo/dynamo-frontend-codeowners @ai-dynamo/dynamo-observability-codeowners
+/lib/llm/src/kv_router/shared_cache/                                                                    @ai-dynamo/dynamo-router-codeowners @ai-dynamo/dynamo-kv-memory-codeowners
 /container/deps/requirements.vllm.txt                                                                   @ai-dynamo/dynamo-ops-codeowners @ai-dynamo/dynamo-backend-vllm-codeowners
 /deploy/operator/internal/checkpoint/                                                                   @ai-dynamo/dynamo-gms-codeowners @ai-dynamo/dynamo-operator-codeowners
 /lib/llm/src/preprocessor/mm_routing/                                                                   @ai-dynamo/dynamo-frontend-codeowners @ai-dynamo/dynamo-multimodal-codeowners @ai-dynamo/dynamo-router-codeowners

components/src/dynamo/common/configuration/groups/kv_router_args.py

@@ -235,6 +235,10 @@ class KvRouterConfigBase(ConfigBase):
     def apply_load_aware_preset(self) -> None:
         if not self.load_aware:
             return
+        if self.shared_cache_type == "mooncake-store":
+            raise ValueError(
+                "--shared-cache-type=mooncake-store is incompatible with --load-aware"
+            )
 
         for field, value in _LOAD_AWARE_KWARG_OVERRIDES.items():
             setattr(self, field, value)
@@ -693,11 +697,15 @@ def add_arguments(self, parser) -> None:
             help=(
                 "[EXPERIMENTAL] KV Router: Type of external shared KV cache to query. "
                 "'none' (default): disabled. "
-                "'hicache': query Mooncake master directly for SGLang L3 (HiCache) state "
-                "using SGLang-compatible Mooncake key derivation."
+                "'hicache': index Mooncake events using SGLang HiCache keys. "
+                "'mooncake-store': index vLLM MooncakeStoreConnector objects using "
+                "resolved worker metadata and GPU events. Requires the explicit "
+                "MooncakeStoreWorkerExtension, compatible pinned dependencies, "
+                "local KV events, and positive device-overlap credit; "
+                "not compatible with --use-remote-indexer."
             ),
             arg_type=str,
-            choices=["none", "hicache"],
+            choices=["none", "hicache", "mooncake-store"],
         )
         add_argument(
             g,

components/src/dynamo/common/tests/configuration/test_kv_router_args.py

@@ -181,6 +181,127 @@ def test_deprecated_overlap_score_weight_env_coexists_with_canonical_settings(
     assert config.kv_router_kwargs()["overlap_score_weight"] == 0.0
 
 
+@pytest.mark.parametrize("cache_type", ["none", "hicache", "mooncake-store"])
+def test_shared_cache_type_cli_flows_to_binding_kwargs(cache_type: str) -> None:
+    parser = argparse.ArgumentParser()
+    KvRouterArgGroup().add_arguments(parser)
+
+    args = parser.parse_args(["--shared-cache-type", cache_type])
+    kwargs = KvRouterConfigBase.from_cli_args(args).kv_router_kwargs()
+
+    assert kwargs["shared_cache_type"] == cache_type
+    assert kwargs["shared_cache_multiplier"] == 0.5
+
+
+def test_mooncake_store_environment_and_cli_precedence(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("DYN_SHARED_CACHE_TYPE", "mooncake-store")
+    monkeypatch.setenv("DYN_SHARED_CACHE_MULTIPLIER", "0.3")
+    parser = argparse.ArgumentParser()
+    KvRouterArgGroup().add_arguments(parser)
+
+    kwargs = KvRouterConfigBase.from_cli_args(parser.parse_args([])).kv_router_kwargs()
+    assert kwargs["shared_cache_type"] == "mooncake-store"
+    assert kwargs["shared_cache_multiplier"] == 0.3
+
+    args = parser.parse_args(
+        ["--shared-cache-type", "none", "--shared-cache-multiplier", "0"]
+    )
+    kwargs = KvRouterConfigBase.from_cli_args(args).kv_router_kwargs()
+    assert kwargs["shared_cache_type"] == "none"
+    assert kwargs["shared_cache_multiplier"] == 0.0
+
+
+@pytest.mark.parametrize("cache_type", ["mooncakestore", "mooncake_store", "redis"])
+def test_shared_cache_type_rejects_unknown_cli_values(cache_type: str) -> None:
+    parser = argparse.ArgumentParser()
+    KvRouterArgGroup().add_arguments(parser)
+
+    with pytest.raises(SystemExit):
+        parser.parse_args(["--shared-cache-type", cache_type])
+
+
+@pytest.mark.parametrize(
+    "router_mode",
+    [
+        "round-robin",
+        "random",
+        "power-of-two",
+        "direct",
+        "least-loaded",
+        "device-aware-weighted",
+    ],
+)
+def test_mooncake_store_frontend_requires_kv_router_mode(router_mode: str) -> None:
+    parser = argparse.ArgumentParser()
+    FrontendArgGroup().add_arguments(parser)
+    config = FrontendConfig.from_cli_args(
+        parser.parse_args(
+            ["--shared-cache-type", "mooncake-store", "--router-mode", router_mode]
+        )
+    )
+
+    with pytest.raises(ValueError, match="mooncake-store requires --router-mode=kv"):
+        config.validate()
+    assert config.shared_cache_type == "mooncake-store"
+
+
+@pytest.mark.parametrize("frontend", [False, True])
+def test_mooncake_store_rejects_load_aware_before_preset(frontend: bool) -> None:
+    parser = argparse.ArgumentParser()
+    group = FrontendArgGroup() if frontend else KvRouterArgGroup()
+    group.add_arguments(parser)
+    argv = ["--shared-cache-type", "mooncake-store", "--load-aware"]
+    if frontend:
+        argv.extend(["--router-mode", "kv"])
+    config_class = FrontendConfig if frontend else KvRouterConfigBase
+    config = config_class.from_cli_args(parser.parse_args(argv))
+
+    with pytest.raises(
+        ValueError, match="mooncake-store is incompatible with --load-aware"
+    ):
+        if frontend:
+            config.validate()
+        else:
+            config.kv_router_kwargs()
+    assert config.shared_cache_type == "mooncake-store"
+    assert config.use_kv_events is True
+    assert config.overlap_score_credit == 1.0
+
+
+def test_mooncake_store_frontend_preserves_parsed_config_over_environment(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("DYN_SHARED_CACHE_TYPE", "hicache")
+    monkeypatch.setenv("DYN_ROUTER_MODE", "round-robin")
+    parser = argparse.ArgumentParser()
+    FrontendArgGroup().add_arguments(parser)
+    config = FrontendConfig.from_cli_args(
+        parser.parse_args(
+            ["--shared-cache-type", "mooncake-store", "--router-mode", "kv"]
+        )
+    )
+
+    config.validate()
+    assert config.shared_cache_type == "mooncake-store"
+    assert config.kv_router_kwargs()["shared_cache_type"] == "mooncake-store"
+    assert config.router_mode == "kv"
+    assert kv_router_args.os.environ["DYN_SHARED_CACHE_TYPE"] == "hicache"
+    assert kv_router_args.os.environ["DYN_ROUTER_MODE"] == "round-robin"
+
+
+def test_mooncake_store_help_names_worker_and_event_requirements() -> None:
+    parser = argparse.ArgumentParser()
+    KvRouterArgGroup().add_arguments(parser)
+    help_by_dest = {action.dest: action.help for action in parser._actions}
+
+    shared_cache_help = help_by_dest["shared_cache_type"]
+    assert "MooncakeStoreWorkerExtension" in shared_cache_help
+    assert "local KV events" in shared_cache_help
+    assert "--use-remote-indexer" in shared_cache_help
+
+
 def test_decode_active_request_weight_flows_to_binding_kwargs() -> None:
     parser = argparse.ArgumentParser()
     KvRouterArgGroup().add_arguments(parser)

components/src/dynamo/frontend/frontend_args.py

@@ -107,6 +107,10 @@ class FrontendConfig(RouterConfigBase, KvRouterConfigBase, AicPerfConfigBase):
     _VALID_TOKENIZER_BACKENDS = {"default", "fastokens", "basetenkenizer"}
 
     def validate(self) -> None:
+        if self.shared_cache_type == "mooncake-store" and self.router_mode != "kv":
+            raise ValueError(
+                "--shared-cache-type=mooncake-store requires --router-mode=kv"
+            )
         if self.load_aware:
             self.router_mode = "kv"
         self.apply_load_aware_preset()

components/src/dynamo/vllm/args.py

@@ -411,7 +411,16 @@ def update_engine_config_with_dynamo(
                     "worker_extension_cls=%s",
                     worker_extension_cls,
                 )
-            elif str(existing_ext) != worker_extension_cls:
+            elif existing_ext == (
+                "dynamo.vllm.mooncake_store_runtime.MooncakeStoreWorkerExtension"
+            ):
+                defaults[
+                    "worker_extension_cls"
+                ] = "dynamo.vllm.mooncake_store_worker.MooncakeStoreFpmWorkerExtension"
+            elif existing_ext not in (
+                worker_extension_cls,
+                "dynamo.vllm.mooncake_store_worker.MooncakeStoreFpmWorkerExtension",
+            ):
                 raise ValueError(
                     f"DYN_FPM_GC_POLICY requires "
                     f"worker_extension_cls='{worker_extension_cls}' so model "

components/src/dynamo/vllm/main.py

@@ -76,6 +76,7 @@
 from .kv_connector_protocols import (
     disable_hybrid_kv_cache_manager_for_incompatible_pd_connector,
 )
+from .mooncake_store_runtime import publish_mooncake_store_runtime
 from .multimodal_utils.cache_config import configure_multimodal_embedding_cache
 from .multimodal_utils.media_config import create_frontend_media_config
 from .multimodal_utils.models.nemotron_video_routing import (
@@ -875,6 +876,13 @@ async def register_vllm_model(
     )
     runtime_values = get_engine_cache_info(engine_client)
     num_gpu_blocks = runtime_values["num_gpu_blocks"]
+    await publish_mooncake_store_runtime(
+        runtime_config,
+        engine_client,
+        vllm_config,
+        dp_range=dp_range,
+        event_span=runtime_values["kv_event_block_size"],
+    )
     # Get data_parallel_size from vllm_config (defaults to 1)
     if num_gpu_blocks is None:
         # TODO(upstream-vllm): remove this workaround once vLLM propagates