feat(router): add GMS placement protocol surfaces

Signed-off-by: Maksim Khadkevich <mkhadkevich@nvidia.com>

Author: hutm

Cargo.lock

@@ -36,6 +36,7 @@
     "router_track_output_blocks",
     "router_assume_kv_reuse",
     "router_track_prefill_tokens",
+    "router_gms_decode_transfer",
     "router_prefill_load_model",
     "router_snapshot_threshold",
     "router_reset_states",
@@ -116,6 +117,7 @@ class KvRouterConfigBase(ConfigBase):
     router_track_output_blocks: bool
     router_assume_kv_reuse: bool
     router_track_prefill_tokens: bool
+    router_gms_decode_transfer: bool
     router_prefill_load_model: str
     router_snapshot_threshold: int
     router_reset_states: bool
@@ -296,6 +298,18 @@ def add_arguments(self, parser) -> None:
                 "prefill-token load, queue pressure, and active_prefill_tokens metrics."
             ),
         )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--router-gms-decode-transfer",
+            env_var="DYN_ROUTER_GMS_DECODE_TRANSFER",
+            default=False,
+            dest="router_gms_decode_transfer",
+            help=(
+                "[EXPERIMENTAL] KV Router: Enable GMS decode-to-decode KV transfer "
+                "orchestration for GMS-capable aggregated decode workers. Disabled "
+                "by default so vanilla Dynamo routing behavior is unchanged."
+            ),
+        )
         add_argument(
             g,
             flag_name="--router-prefill-load-model",

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

@@ -414,6 +414,78 @@ def test_load_aware_frontend_implies_kv_router_mode() -> None:
     assert config.router_assume_kv_reuse is False
 
 
+def test_bulwark_gateway_endpoint_requires_private_namespace() -> None:
+    parser = argparse.ArgumentParser()
+    FrontendArgGroup().add_arguments(parser)
+
+    args = parser.parse_args(
+        ["--bulwark-gateway-endpoint", "dyn://public.backend.generate"]
+    )
+    config = FrontendConfig.from_cli_args(args)
+
+    with pytest.raises(ValueError, match="requires --namespace"):
+        config.validate()
+
+
+def test_bulwark_gateway_endpoint_rejects_non_endpoint_path() -> None:
+    parser = argparse.ArgumentParser()
+    FrontendArgGroup().add_arguments(parser)
+
+    args = parser.parse_args(
+        [
+            "--bulwark-gateway-endpoint",
+            "public.backend.generate",
+            "--namespace",
+            "private-workers",
+        ]
+    )
+    config = FrontendConfig.from_cli_args(args)
+
+    with pytest.raises(ValueError, match="dyn://"):
+        config.validate()
+
+
+def test_bulwark_gateway_endpoint_validates() -> None:
+    parser = argparse.ArgumentParser()
+    FrontendArgGroup().add_arguments(parser)
+
+    args = parser.parse_args(
+        [
+            "--bulwark-gateway-endpoint",
+            "dyn://public.backend.generate",
+            "--namespace-prefix",
+            "private-workers",
+        ]
+    )
+    config = FrontendConfig.from_cli_args(args)
+    config.validate()
+
+    assert config.bulwark_gateway_endpoint == "dyn://public.backend.generate"
+    assert config.namespace_prefix == "private-workers"
+
+
+def test_bulwark_gateway_exact_namespace_uses_worker_suffix(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("DYN_NAMESPACE_WORKER_SUFFIX", "replica-a")
+    parser = argparse.ArgumentParser()
+    FrontendArgGroup().add_arguments(parser)
+
+    args = parser.parse_args(
+        [
+            "--bulwark-gateway-endpoint",
+            "dyn://public.backend.generate",
+            "--namespace",
+            "private-workers",
+        ]
+    )
+    config = FrontendConfig.from_cli_args(args)
+    config.validate()
+
+    assert config.namespace == "private-workers-replica-a"
+    assert config.namespace_prefix is None
+
+
 def test_frontend_admission_control_defaults_to_none(monkeypatch) -> None:
     """Default --admission-control is 'none': busy thresholds are cleared
     even though the underlying threshold flags have non-None defaults."""

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

@@ -25,6 +25,7 @@
     add_negatable_bool_argument,
     env_or_default,
 )
+from dynamo.common.utils.namespace import get_worker_namespace
 
 from . import __version__
 
@@ -95,6 +96,8 @@ def validate(self) -> None:
         self.apply_load_aware_preset()
 
         if self.bulwark_gateway_endpoint:
+            if self.namespace and not self.namespace_prefix:
+                self.namespace = get_worker_namespace(self.namespace)
             if not self.bulwark_gateway_endpoint.startswith("dyn://"):
                 raise ValueError(
                     "--bulwark-gateway-endpoint must be a dyn://namespace.component.endpoint path"

components/src/dynamo/frontend/frontend_args.py

@@ -680,6 +680,9 @@ fn kv_router_config_from_env() -> KvRouterConfig {
     if let Some(v) = env_bool("DYN_ROUTER_TRACK_PREFILL_TOKENS") {
         cfg.router_track_prefill_tokens = v;
     }
+    if let Some(v) = env_bool("DYN_ROUTER_GMS_DECODE_TRANSFER") {
+        cfg.router_gms_decode_transfer = v;
+    }
     if let Some(v) = env_f64("DYN_ROUTER_QUEUE_THRESHOLD") {
         cfg.router_queue_threshold = Some(v);
     }
@@ -695,6 +698,7 @@ fn kv_router_config_from_env() -> KvRouterConfig {
         router_track_active_blocks = cfg.router_track_active_blocks,
         router_track_output_blocks = cfg.router_track_output_blocks,
         router_track_prefill_tokens = cfg.router_track_prefill_tokens,
+        router_gms_decode_transfer = cfg.router_gms_decode_transfer,
         router_queue_threshold = ?cfg.router_queue_threshold,
         router_predicted_ttl_secs = ?cfg.router_predicted_ttl_secs,
         queue_depth_tiers_unbounded = cfg.router_queue_by_incoming_missing_isl.is_unbounded(),

lib/bindings/c/src/lib.rs

@@ -175,7 +175,7 @@ impl AicPerfConfig {
 #[pymethods]
 impl KvRouterConfig {
     #[new]
-    #[pyo3(signature = (overlap_score_weight=None, host_cache_hit_weight=0.75, disk_cache_hit_weight=0.25, router_temperature=0.0, use_kv_events=true, durable_kv_events=false, router_replica_sync=false, router_track_active_blocks=true, router_track_output_blocks=false, router_assume_kv_reuse=true, router_track_prefill_tokens=true, router_prefill_load_model="none", router_snapshot_threshold=1000000, router_reset_states=false, router_ttl_secs=120.0, router_queue_threshold=Some(16.0), router_event_threads=4, router_queue_policy="fcfs", use_remote_indexer=false, serve_indexer=false, shared_cache_multiplier=0.0, shared_cache_type="none", router_predicted_ttl_secs=None, *, overlap_score_credit=1.0, prefill_load_scale=1.0, router_queue_by_incoming_missing_isl=None))]
+    #[pyo3(signature = (overlap_score_weight=None, host_cache_hit_weight=0.75, disk_cache_hit_weight=0.25, router_temperature=0.0, use_kv_events=true, durable_kv_events=false, router_replica_sync=false, router_track_active_blocks=true, router_track_output_blocks=false, router_assume_kv_reuse=true, router_track_prefill_tokens=true, router_prefill_load_model="none", router_snapshot_threshold=1000000, router_reset_states=false, router_ttl_secs=120.0, router_queue_threshold=Some(16.0), router_event_threads=4, router_queue_policy="fcfs", use_remote_indexer=false, serve_indexer=false, shared_cache_multiplier=0.0, shared_cache_type="none", router_predicted_ttl_secs=None, *, overlap_score_credit=1.0, prefill_load_scale=1.0, router_queue_by_incoming_missing_isl=None, router_gms_decode_transfer=false))]
     #[allow(clippy::too_many_arguments)]
     fn new(
         overlap_score_weight: Option<f64>,
@@ -204,6 +204,7 @@ impl KvRouterConfig {
         mut overlap_score_credit: f64,
         mut prefill_load_scale: f64,
         router_queue_by_incoming_missing_isl: Option<Vec<(usize, usize)>>,
+        router_gms_decode_transfer: bool,
     ) -> PyResult<Self> {
         if let Some(value) = overlap_score_weight {
             apply_deprecated_overlap_score_weight(
@@ -226,6 +227,7 @@ impl KvRouterConfig {
             router_track_output_blocks,
             router_assume_kv_reuse,
             router_track_prefill_tokens,
+            router_gms_decode_transfer,
             router_prefill_load_model: router_prefill_load_model
                 .parse::<RsRouterPrefillLoadModel>()
                 .map_err(PyValueError::new_err)?,

lib/bindings/python/rust/llm/entrypoint.rs

@@ -370,6 +370,58 @@ impl KvEventPublisher {
         })
     }
 
+    #[pyo3(signature = (source_nixl_agent_name, source_nixl_agent_metadata_hex, blocks, source_nixl_ip=None, source_nixl_listen_port=None, dp_rank=None))]
+    #[allow(clippy::too_many_arguments)]
+    fn publish_gms_placement_stored(
+        &self,
+        py: Python,
+        source_nixl_agent_name: String,
+        source_nixl_agent_metadata_hex: String,
+        blocks: Bound<PyAny>,
+        source_nixl_ip: Option<String>,
+        source_nixl_listen_port: Option<u16>,
+        dp_rank: Option<DpRank>,
+    ) -> PyResult<()> {
+        let blocks: Vec<GmsPlacementBlock> = depythonize(&blocks).map_err(to_pyerr)?;
+        let inner = self.inner.clone();
+        let dp_rank = dp_rank.unwrap_or(self.dp_rank);
+        py.allow_threads(|| {
+            inner
+                .publish_gms_placement(
+                    GmsPlacementEventData::Stored(GmsPlacementStoreData {
+                        source_nixl_agent_name,
+                        source_nixl_agent_metadata_hex,
+                        source_nixl_ip,
+                        source_nixl_listen_port,
+                        blocks,
+                    }),
+                    StorageTier::External,
+                    dp_rank,
+                )
+                .map_err(to_pyerr)
+        })
+    }
+
+    #[pyo3(signature = (content_hashes_hex, dp_rank=None))]
+    fn publish_gms_placement_removed(
+        &self,
+        py: Python,
+        content_hashes_hex: Vec<String>,
+        dp_rank: Option<DpRank>,
+    ) -> PyResult<()> {
+        let inner = self.inner.clone();
+        let dp_rank = dp_rank.unwrap_or(self.dp_rank);
+        py.allow_threads(|| {
+            inner
+                .publish_gms_placement(
+                    GmsPlacementEventData::Removed(GmsPlacementRemoveData { content_hashes_hex }),
+                    StorageTier::External,
+                    dp_rank,
+                )
+                .map_err(to_pyerr)
+        })
+    }
+
     fn shutdown(&mut self) {
         // If no other Arc clones exist, shut down eagerly.
         // Otherwise the Drop impl handles cleanup when the last reference is freed.

lib/bindings/python/rust/llm/kv.rs

@@ -159,6 +159,20 @@ impl ModelRuntimeConfig {
         Ok(())
     }
 
+    fn set_gms_placement_enabled(&mut self) -> PyResult<()> {
+        self.inner.set_gms_placement_enabled().map_err(to_pyerr)
+    }
+
+    fn set_gms_daemon_socket(&mut self, daemon_socket: String) -> PyResult<()> {
+        self.inner
+            .set_gms_daemon_socket(daemon_socket)
+            .map_err(to_pyerr)
+    }
+
+    fn set_gms_control_enabled(&mut self) -> PyResult<()> {
+        self.inner.set_gms_control_enabled().map_err(to_pyerr)
+    }
+
     fn set_tensor_model_config(
         &mut self,
         _py: Python<'_>,

lib/bindings/python/rust/llm/local_model.rs

@@ -712,6 +712,18 @@ class ModelRuntimeConfig:
         """Set structural tag schema mode ("auto" or "strict")."""
         ...
 
+    def set_gms_daemon_socket(self, daemon_socket: str) -> None:
+        """Advertise this worker's GMS KV daemon socket for router orchestration"""
+        ...
+
+    def set_gms_placement_enabled(self) -> None:
+        """Advertise that this worker publishes GMS placements on the KV event plane"""
+        ...
+
+    def set_gms_control_enabled(self) -> None:
+        """Deprecated alias for set_gms_placement_enabled."""
+        ...
+
     def set_disaggregated_endpoint(
             self,
             bootstrap_host: str | None = None,
@@ -1032,6 +1044,26 @@ class KvEventPublisher:
         """
         ...
 
+    def publish_gms_placement_stored(
+        self,
+        source_nixl_agent_name: str,
+        source_nixl_agent_metadata_hex: str,
+        blocks: List[Dict[str, Any]],
+        source_nixl_ip: Optional[str] = None,
+        source_nixl_listen_port: Optional[int] = None,
+        dp_rank: Optional[int] = None,
+    ) -> None:
+        """Publish GMS placement descriptors through the KV event plane."""
+        ...
+
+    def publish_gms_placement_removed(
+        self,
+        content_hashes_hex: List[str],
+        dp_rank: Optional[int] = None,
+    ) -> None:
+        """Remove GMS placement descriptors from the router's event-plane index."""
+        ...
+
     def shutdown(self) -> None:
         """
         Shuts down the event publisher, stopping any background tasks.
@@ -1673,6 +1705,7 @@ class KvRouterConfig:
         overlap_score_credit: float = 1.0,
         prefill_load_scale: float = 1.0,
         router_queue_by_incoming_missing_isl: Optional[list[tuple[int, int]]] = None,
+        router_gms_decode_transfer: bool = False,
     ) -> None:
         """
         Create a KV router configuration.
@@ -1697,6 +1730,7 @@ class KvRouterConfig:
             router_assume_kv_reuse: Assume KV cache reuse when tracking active blocks (default: True).
                 When True, computes actual block hashes. When False, generates random hashes.
             router_track_prefill_tokens: Include prompt-side prefill tokens in active load accounting (default: True).
+            router_gms_decode_transfer: Enable experimental GMS decode-to-decode transfer orchestration (default: False).
             router_prefill_load_model: Prompt-side prefill load model (default: "none").
                 "none" keeps static prompt load accounting.
                 "aic" decays the oldest active prefill request using AIC-predicted duration.

lib/bindings/python/src/dynamo/_core.pyi

@@ -63,4 +63,5 @@ rstest_reuse = "0.7.0"
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["rt", "macros", "time", "test-util"] }
 dynamo-tokens = { workspace = true }
+bincode = { version = "1" }
 tower = { version = "0.5", features = ["util"] }