feat(runtime): add endpoint-scoped event transport [DYN-3580]

components/src/dynamo/common/backend/worker.py

@@ -144,6 +144,8 @@ class WorkerConfig:
     route_to_encoder: bool = False
     media_decoder: Optional[MediaDecoder] = None
     media_fetcher: Optional[MediaFetcher] = None
+    # KV event/recovery ownership endpoint. None uses this worker's serving endpoint.
+    kv_state_endpoint: Optional[str] = None
 
     @classmethod
     def from_runtime_config(
@@ -163,6 +165,7 @@ def from_runtime_config(
             "namespace": runtime_cfg.namespace,
             "component": getattr(runtime_cfg, "component", None) or "backend",
             "endpoint": getattr(runtime_cfg, "endpoint", None) or "generate",
+            "kv_state_endpoint": getattr(runtime_cfg, "kv_state_endpoint", None),
             "model_name": model_name,
             "served_model_name": served_model_name,
             "endpoint_types": getattr(
@@ -255,6 +258,7 @@ async def run(self) -> None:
             namespace=self.config.namespace,
             component=self.config.component,
             endpoint=self.config.endpoint,
+            kv_state_endpoint=self.config.kv_state_endpoint,
             model_name=self.config.model_name,
             served_model_name=self.config.served_model_name,
             model_input=self.config.model_input,

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

@@ -25,6 +25,9 @@ class DynamoRuntimeConfig(ConfigBase):
 
     namespace: str
     endpoint: Optional[str] = None
+    # Exact endpoint that owns this worker's KV event and recovery state. None
+    # maps KV state to the serving endpoint and does not change request routing.
+    kv_state_endpoint: Optional[str] = None
     discovery_backend: str
     request_plane: str
     event_plane: Optional[str] = None
@@ -128,6 +131,13 @@ def add_arguments(self, parser: argparse.ArgumentParser) -> None:
             default=None,
             help="Dynamo endpoint string in 'dyn://namespace.component.endpoint' format. Example: dyn://dynamo.backend.generate.",
         )
+        add_argument(
+            g,
+            flag_name="--kv-state-endpoint",
+            env_var="DYN_KV_STATE_ENDPOINT",
+            default=None,
+            help="Endpoint identity that owns this worker's KV event and recovery state. Defaults to the serving endpoint.",
+        )
         add_argument(
             g,
             flag_name="--discovery-backend",

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

@@ -40,6 +40,17 @@ def test_fpm_trace_defaults_disabled(monkeypatch):
     assert "DYN_FPM_TRACE" not in os.environ
 
 
+def test_kv_state_endpoint_supports_cli_and_env(monkeypatch):
+    monkeypatch.setenv("DYN_KV_STATE_ENDPOINT", "dynamo/kv/events")
+    env_config, help_text = _parse_runtime_args([])
+    cli_config, _ = _parse_runtime_args(["--kv-state-endpoint", "other/cache/updates"])
+
+    assert env_config.kv_state_endpoint == "dynamo/kv/events"
+    assert cli_config.kv_state_endpoint == "other/cache/updates"
+    assert "--kv-state-endpoint" in help_text
+    assert "DYN_KV_STATE_ENDPOINT" in help_text
+
+
 def test_fpm_trace_env_enables_and_is_canonicalized(monkeypatch):
     monkeypatch.setenv("DYN_FPM_TRACE", "on")
 

components/src/dynamo/planner/tests/integration/test_fpm_event_delivery.py

@@ -0,0 +1,215 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+from contextlib import AsyncExitStack
+
+import pytest
+
+from dynamo.llm import (
+    FpmDirectPublisher,
+    ModelInput,
+    ModelType,
+    WorkerType,
+    register_model,
+)
+from dynamo.planner.config.planner_config import PlannerConfig
+from dynamo.planner.core.base import NativePlannerBase
+from dynamo.planner.core.state_machine import PlannerScalingState
+from dynamo.planner.core.types import (
+    EngineCapabilities,
+    ScheduledTick,
+    WorkerCapabilities,
+    WorkerCounts,
+)
+from dynamo.planner.environment.metrics_provider.runtime_provider import (
+    RuntimeFpmProvider,
+)
+from dynamo.planner.environment.state import DeploymentState
+from dynamo.planner.monitoring.worker_info import WorkerInfo
+from dynamo.runtime import DistributedRuntime
+
+pytestmark = [
+    pytest.mark.gpu_0,
+    pytest.mark.pre_merge,
+    pytest.mark.integration,
+    pytest.mark.planner,
+]
+
+
+class _PlannerSources:
+    def __init__(self, namespace: str, state: DeploymentState) -> None:
+        self._namespace = namespace
+        self._state = state
+
+    def deployment_state(self) -> DeploymentState:
+        return self._state
+
+    def runtime_namespace(self) -> str:
+        return self._namespace
+
+
+class _PlannerEnvironment(_PlannerSources):
+    def __init__(self, namespace, state, provider):
+        super().__init__(namespace, state)
+        self._provider = provider
+
+    def collect_fpm(self):
+        return self._provider.collect_fpm()
+
+
+class _DecodePlanner(NativePlannerBase):
+    require_decode = True
+
+
+@pytest.mark.asyncio
+async def test_endpoint_fpm_reaches_planner_and_drives_scale_up(tmp_path, monkeypatch):
+    monkeypatch.setenv("DYN_FILE_KV", str(tmp_path))
+    namespace = "fpm-test"
+    endpoint_path = f"{namespace}.worker.generate"
+    async with AsyncExitStack() as stack:
+        worker_runtime = DistributedRuntime(
+            asyncio.get_running_loop(), "file", "tcp", event_plane="zmq"
+        )
+        stack.callback(worker_runtime.shutdown)
+        planner_runtime = DistributedRuntime(
+            asyncio.get_running_loop(), "file", "tcp", event_plane="zmq"
+        )
+        stack.callback(planner_runtime.shutdown)
+        decoy_runtime = DistributedRuntime(
+            asyncio.get_running_loop(), "file", "tcp", event_plane="zmq"
+        )
+        stack.callback(decoy_runtime.shutdown)
+        worker_endpoint = worker_runtime.endpoint(endpoint_path)
+        worker_id = str(worker_endpoint.connection_id())
+        decoy_endpoint = decoy_runtime.endpoint(f"{namespace}.worker.generate-decoy")
+        # Deliberately reuse the serving worker ID. If endpoint scoping regresses, identity-based
+        # assertions cannot distinguish this decoy from the configured serving pool.
+        decoy_publisher = FpmDirectPublisher(decoy_endpoint, worker_id, dp_size=1)
+        stack.callback(decoy_publisher.shutdown)
+        await register_model(
+            ModelInput.Tensor,
+            ModelType.TensorBased,
+            worker_endpoint,
+            "test-model",
+            worker_type=WorkerType.Aggregated,
+        )
+        state = DeploymentState()
+        state.decode.info = WorkerInfo(
+            component_name="worker",
+            endpoint="generate",
+            model_name="test-model",
+        )
+        state.decode.replicas.active = 1
+        state.decode.replicas.expected = 1
+        sources = _PlannerSources(namespace, state)
+        config = PlannerConfig(
+            mode="agg",
+            enable_load_scaling=True,
+            enable_throughput_scaling=False,
+            optimization_target="load",
+            served_model_name="test-model",
+            decode_scale_up_kv_rate=90.0,
+            decode_scale_down_kv_rate=50.0,
+            min_endpoint=1,
+            max_gpu_budget=-1,
+            metric_reporting_prometheus_port=0,
+        )
+        # Establish a decoy-only quiet phase before the serving publisher emits anything. A
+        # component-scoped subscription would observe this as the real worker because the IDs are
+        # intentionally identical; the endpoint-scoped subscriber must remain empty.
+        async with AsyncExitStack() as isolation_stack:
+            isolation_provider = RuntimeFpmProvider(
+                require_prefill=False,
+                require_decode=True,
+                backend="vllm",
+                model_name="test-model",
+                runtime=planner_runtime,
+                state_source=sources,
+                namespace_source=sources,
+            )
+            await isolation_provider.async_init()
+            isolation_stack.push_async_callback(isolation_provider.shutdown)
+            for _ in range(20):
+                decoy_publisher.publish(
+                    dp_rank=0,
+                    scheduled_num_prefill_requests=0,
+                    scheduled_sum_prefill_tokens=0,
+                    scheduled_sum_prefill_kv_tokens=0,
+                    scheduled_num_decode_requests=1,
+                    scheduled_sum_decode_kv_tokens=999,
+                    queued_num_prefill_requests=0,
+                    queued_sum_prefill_tokens=0,
+                    queued_num_decode_requests=1,
+                    queued_sum_decode_kv_tokens=999,
+                    wall_time_secs=0.01,
+                )
+                await asyncio.sleep(0.05)
+                assert not isolation_provider.collect_fpm().decode
+
+        publisher = FpmDirectPublisher(worker_endpoint, worker_id, dp_size=1)
+        stack.callback(publisher.shutdown)
+        provider = RuntimeFpmProvider(
+            require_prefill=False,
+            require_decode=True,
+            backend="vllm",
+            model_name="test-model",
+            runtime=planner_runtime,
+            state_source=sources,
+            namespace_source=sources,
+        )
+        await provider.async_init()
+        stack.push_async_callback(provider.shutdown)
+        planner = _DecodePlanner(
+            None,
+            config,
+            _PlannerEnvironment(namespace, state, provider),
+        )
+
+        tick_input = None
+        for _ in range(100):
+            publisher.publish(
+                dp_rank=0,
+                scheduled_num_prefill_requests=0,
+                scheduled_sum_prefill_tokens=0,
+                scheduled_sum_prefill_kv_tokens=0,
+                scheduled_num_decode_requests=1,
+                scheduled_sum_decode_kv_tokens=200,
+                queued_num_prefill_requests=0,
+                queued_sum_prefill_tokens=0,
+                queued_num_decode_requests=1,
+                queued_sum_decode_kv_tokens=100,
+                wall_time_secs=0.01,
+            )
+            await asyncio.sleep(0.05)
+            tick_input = await planner._gather_tick_input(
+                ScheduledTick(
+                    at_s=0.0,
+                    need_worker_states=True,
+                    need_worker_fpm=True,
+                    run_load_scaling=True,
+                )
+            )
+            if tick_input.fpm_observations.decode:
+                break
+
+        assert tick_input is not None
+        observations = tick_input.fpm_observations
+        assert observations.decode is not None
+        assert (worker_id, 0) in observations.decode
+
+        scaling = PlannerScalingState(
+            config,
+            WorkerCapabilities(decode=EngineCapabilities(num_gpu=1, max_kv_tokens=100)),
+        )
+        assert tick_input.worker_counts == WorkerCounts(
+            ready_num_decode=1,
+            expected_num_decode=1,
+        )
+        scaling.observe_worker_counts(tick_input.worker_counts)
+
+        decision = scaling.advance_load(observations)
+
+        assert decision is not None
+        assert decision.num_decode is not None
+        assert decision.num_decode > 1

components/src/dynamo/sglang/publisher.py

@@ -338,6 +338,7 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]:
                     zmq_topic="",
                     enable_local_indexer=self.dynamo_args.enable_local_indexer,
                     dp_rank=dp_rank,
+                    kv_state_endpoint=self.dynamo_args.kv_state_endpoint,
                 )
                 self.kv_publishers.append(publisher)
 

components/src/dynamo/sglang/register.py

@@ -358,6 +358,7 @@ async def _get_runtime_config(
         ModelRuntimeConfig with extracted values, or None if extraction fails.
     """
     runtime_config = ModelRuntimeConfig()
+    runtime_config.kv_state_endpoint = dynamo_args.kv_state_endpoint
     runtime_config.context_length = server_args.context_length
     # set reasoning parser and tool call parser
     runtime_config.reasoning_parser = dynamo_args.dyn_reasoning_parser

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

@@ -404,7 +404,7 @@ def __init__(self, **kwargs):
     )
     config = SimpleNamespace(
         server_args=server_args,
-        dynamo_args=SimpleNamespace(enable_local_indexer=True),
+        dynamo_args=SimpleNamespace(enable_local_indexer=True, kv_state_endpoint=None),
     )
     publisher = DynamoSglangPublisher(
         engine=SimpleNamespace(),
@@ -461,7 +461,7 @@ def shutdown(self):
     )
     config = SimpleNamespace(
         server_args=server_args,
-        dynamo_args=SimpleNamespace(enable_local_indexer=True),
+        dynamo_args=SimpleNamespace(enable_local_indexer=True, kv_state_endpoint=None),
     )
     publisher = DynamoSglangPublisher(
         engine=SimpleNamespace(

components/src/dynamo/trtllm/publisher.py

@@ -375,6 +375,10 @@ class Publisher:
     Note: The ZmqKvEventPublisher used here is the pure Python ZMQ publisher defined
     in this module, not the Rust-based KvEventPublisher from dynamo.llm (which is
     used in main.py as the worker-side subscriber from consolidator to NATS).
+
+    ``kv_state_endpoint`` selects the exact Dynamo endpoint that owns the published
+    KV event and recovery state. ``None`` maps KV state to the serving endpoint; it
+    does not change the endpoint used for request routing.
     """
 
     def __init__(
@@ -390,6 +394,7 @@ def __init__(
         zmq_endpoint: Optional[str] = None,
         enable_local_indexer: bool = False,
         metrics_collector: Any = None,
+        kv_state_endpoint: Optional[str] = None,
     ) -> None:
         self.endpoint = endpoint
         self.engine = engine
@@ -403,6 +408,7 @@ def __init__(
             self.additional_metrics.set_kv_event_buffer_capacity(event_buffer_max_size)
         self.enable_local_indexer = enable_local_indexer
         self.metrics_collector = metrics_collector
+        self.kv_state_endpoint = kv_state_endpoint
         self.attention_dp_size = engine.get_attention_dp_size()
 
         # The first few kv events from the model engine are always "created" type events.
@@ -510,6 +516,7 @@ def initialize(self) -> None:
                     kv_block_size=self.kv_block_size,
                     dp_rank=rank,
                     enable_local_indexer=self.enable_local_indexer,
+                    kv_state_endpoint=self.kv_state_endpoint,
                 )
             logging.info(
                 f"Created {self.attention_dp_size} KV event publisher(s) for attention DP ranks"
@@ -1082,6 +1089,7 @@ async def get_publisher(
     zmq_endpoint: Optional[str] = None,
     enable_local_indexer: bool = False,
     metrics_collector: Any = None,
+    kv_state_endpoint: Optional[str] = None,
 ) -> AsyncGenerator[Publisher, None]:
     publisher = Publisher(
         endpoint,
@@ -1095,6 +1103,7 @@ async def get_publisher(
         zmq_endpoint=zmq_endpoint,
         enable_local_indexer=enable_local_indexer,
         metrics_collector=metrics_collector,
+        kv_state_endpoint=kv_state_endpoint,
     )
     try:
         publisher.initialize()

components/src/dynamo/trtllm/workers/llm_worker.py

@@ -534,6 +534,7 @@ async def init_llm_worker(
         # So for now, we just set the parsers from the config
         # TODO: fix this once we have a better way to get total_kv_blocks
         runtime_config = ModelRuntimeConfig()
+        runtime_config.kv_state_endpoint = config.kv_state_endpoint
         runtime_config.context_length = config.max_seq_len
 
         # Set values from config that are available immediately
@@ -771,6 +772,7 @@ async def init_llm_worker(
                     zmq_endpoint=consolidator_output_connect_endpoint,
                     zmq_topic="",
                     enable_local_indexer=config.enable_local_indexer,
+                    kv_state_endpoint=config.kv_state_endpoint,
                 )
                 logging.info(
                     f"Created worker-side publisher for consolidated events: "
@@ -789,6 +791,7 @@ async def init_llm_worker(
                 zmq_endpoint=trtllm_zmq_bind_endpoint,
                 enable_local_indexer=config.enable_local_indexer,
                 metrics_collector=metrics_collector,
+                kv_state_endpoint=config.kv_state_endpoint,
             ) as publisher:
                 handler_config.publisher = publisher
                 handler = RequestHandlerFactory().get_request_handler(handler_config)