fix(sglang): relay multinode DP sidecar KV events

docs/sidecars.md

@@ -36,7 +36,15 @@ dynamo:
 
 The default sidecar commands are `python3 -m dynamo.sglang.sidecar`, `python3 -m dynamo.vllm.sidecar`, and `python3 -m dynamo.trtllm.sidecar`. All three use the shared `--grpc-endpoint` flag.
 
-SGLang exposes gRPC and starts the sidecar only on an endpoint leader; distributed followers are engine-only. srtctl also adds `--incremental-streaming-output` to every SGLang sidecar engine (and logs that it did): the sidecar treats each gRPC chunk as a delta, and without the flag current SGLang builds stream the cumulative text per chunk, which shows up as repeated prefixes in responses and inflated token counts. Set `incremental-streaming-output` in the role's `args` yourself to override. TensorRT-LLM supports sidecars for aggregated workers only and runs the sidecar on MPI rank zero. `dynamo.sidecar_context_length` can override the TRT-LLM context length inferred from `roles.agg.args.max_seq_len`.
+TensorRT-LLM supports sidecars for aggregated workers only and runs the sidecar on MPI rank zero. `dynamo.sidecar_context_length` can override the TRT-LLM context length inferred from `roles.agg.args.max_seq_len`.
+
+## SGLang topology and launch requirements
+
+SGLang's endpoint leader runs the serving sidecar and registers the complete global DP range. With attention DP and `roles.<role>.kv_events: true`, srtctl also exposes native gRPC and runs a sidecar on each follower that owns a KV publisher. Dynamo automatically selects serving or telemetry-only mode from the local engine's `GetServerInfo` metadata; no mode flag is passed. These followers relay their node-local events with the leader's serving-worker identity and the source's global DP rank; they do not add inference workers. Followers without publishers, including TP-only ranks and later pipeline stages, remain engine-only. For example, TP8/DP2 across four nodes publishes on nodes 0 and 2, so only node 2 needs a telemetry sidecar. Each sidecar discovers its local sources through SGLang's `GetServerInfo`; no context file or manually assigned DP rank is needed.
+
+The [multinode attention-DP example](https://github.com/NVIDIA/srt-slurm/blob/main/examples/features/sglang-sidecar-multinode-dp.yaml) uses one logical aggregate worker across two one-GPU nodes, with a custom prebuilt image and KV routing. Set the image path before submitting. Restart the entire distributed engine group and its sidecars together after a failure; the follower relay does not provide independent sidecar restart or event replay recovery.
+
+The [disaggregated multinode attention-DP example](https://github.com/NVIDIA/srt-slurm/blob/main/examples/features/sglang-sidecar-multinode-disagg-dp.yaml) uses four one-GPU nodes: one TP2/DP2 prefill worker on the first pair and one TP2/DP2 decode worker on the second pair. Both leaders run serving sidecars. Prefill enables KV events and a follower telemetry relay; decode keeps prefix caching disabled by default and sets `kv_events: false`, so its follower remains engine-only. KV routing can reuse prefixes on either prefill DP rank. Decode prefix-cache affinity is not enabled or implied. srtctl supplies the role, distributed rendezvous, prefill bootstrap port, and the prefill leader's routable `--bootstrap-host`; the recipe selects NIXL transfer on both roles. The image must include compatible Dynamo and SGLang multinode sidecar support, SGLang's prebuilt native gRPC extension, and NIXL.
 
 ## vLLM topology and launch requirements
 

examples/README.md

@@ -1,6 +1,6 @@
 # Examples
 
-Small, runnable starting points, one per frontend and topology. Every example serves the same model (Qwen3-0.6B) on one node so the files differ only in the frontend and the prefill/decode layout, and a full matrix run finishes in minutes. They are not performance claims. Copy one, change the model, GPU type, topology, and engine flags to match your target, then `srtctl dry-run -f <config>` before submitting.
+Small starting points, one per frontend and topology. The matrix examples serve the same model (Qwen3-0.6B) on one node so the files differ only in the frontend and the prefill/decode layout, and a full matrix run finishes in minutes. The feature examples include additional topologies and build prerequisites. They are not performance claims. Copy one, change the model, GPU type, topology, and engine flags to match your target, then `srtctl dry-run -f <config>` before submitting.
 
 ## Matrix
 
@@ -24,6 +24,8 @@ Every example is written in the 2.0 layout: `engine:` names the engine (a string
 
 | File | Shows |
 | --- | --- |
+| [features/sglang-sidecar-multinode-dp.yaml](features/sglang-sidecar-multinode-dp.yaml) | One SGLang aggregate worker across two one-GPU nodes, global TP2/attention DP2, a serving leader sidecar and a telemetry-only follower, with KV routing |
+| [features/sglang-sidecar-multinode-disagg-dp.yaml](features/sglang-sidecar-multinode-disagg-dp.yaml) | One TP2/attention DP2 prefill worker and one TP2/attention DP2 decode worker across four one-GPU nodes, with a prefill telemetry relay, NIXL transfer, and KV routing |
 | `features/sweep.yaml` | `sweep:` plus `{placeholder}` substitution; one job per combination |
 | `features/override.yaml` | `base` plus `override_*` and `zip_override_*` variants in one file |
 | `features/profiling.yaml` | `profiling:` torch capture on an aggregated worker |
@@ -40,6 +42,8 @@ Every example is written in the 2.0 layout: `engine:` names the engine (a string
 | `features/worker-restart.yaml` | `roles.<role>.restart:` relaunching a killed worker in place with backoff; serves with `benchmark.type: manual` so you can SIGKILL a worker step and watch the sweep log |
 | `features/sglang-weight-cache.yaml` | SGLang fast engine recovery on the existing machinery: a generic per-worker service runs SGLang's weight cache daemon (tensors kept in HBM, handed out over CUDA IPC), the engines start with `--weight-cache-mode client` and skip the disk load, and `roles.<role>.restart` relaunches a killed engine in place. Needs SGLang v0.5.19+; see [docs/sglang-weight-cache.md](../docs/sglang-weight-cache.md) |
 
+See [Native sidecar mode](../docs/config-reference.md#native-sidecar-mode) for configuration and lifecycle behavior.
+
 ## Cluster aliases
 
 The examples reference two kinds of aliases that `srtslurm.yaml` resolves:
@@ -59,8 +63,6 @@ containers:
   smg: /path/to/smg.sqsh                    # Shepherd Model Gateway image (lightseekorg/smg), the router for the smg examples
 ```
 
-`resources.gpu_type` and `gpus_per_node` are set to `h100` and `8`; change them to match the partition you submit to.
-
 ## Optional GPU temperature
 
 To retain temperature with the existing power collector, add this telemetry block

examples/features/sglang-sidecar-multinode-disagg-dp.yaml

@@ -0,0 +1,85 @@
+# Two disaggregated SGLang attention-DP workers across four physical nodes.
+# Each role is one TP2/DP2 worker spanning two one-GPU nodes. Prefill has a
+# serving leader sidecar and a telemetry-only follower relaying KV events.
+# Dynamo selects each sidecar's mode from its local engine metadata.
+# Decode keeps prefix caching disabled: its leader serves inference and its
+# follower runs only the engine. KV routing reuses cached prefill prefixes;
+# this example does not enable experimental decode prefix caching.
+#
+# Supply an image with compatible Dynamo/SGLang multinode sidecar support,
+# SGLang's prebuilt native gRPC extension, and the NIXL transfer backend.
+# Restart each distributed engine group and its sidecars together after failure;
+# independent sidecar restart/replay recovery is unsupported.
+
+schema: 2
+name: "qwen3-0.6b-sglang-sidecar-multinode-disagg-dp"
+
+slurm:
+  time_limit: "00:30:00"
+
+model:
+  path: "hf:Qwen/Qwen3-0.6B"
+  container: "/path/to/sglang-dynamo-multinode-sidecar.sqsh"
+  precision: "bf16"
+
+resources:
+  gpu_type: "h100"
+  gpus_per_node: 1
+
+dynamo:
+  install: false  # Dynamo is preinstalled in the image.
+  request_plane: "nats"
+  event_plane: "nats"
+
+frontend:
+  type: dynamo
+  enable_multiple_frontends: false
+  args:
+    router-mode: "kv"
+
+engine: sglang
+roles:
+  prefill:
+    nodes: 2
+    workers: 1
+    gpus: 2  # Global GPU count for this one distributed prefill worker.
+    sidecar: true
+    kv_events: true
+    env:
+      PYTHONUNBUFFERED: "1"
+    args:
+      served-model-name: "Qwen/Qwen3-0.6B"
+      tensor-parallel-size: 2
+      data-parallel-size: 2
+      enable-dp-attention: true
+      disaggregation-transfer-backend: nixl
+      page-size: 64
+      context-length: 2048
+      max-running-requests: 8
+      mem-fraction-static: 0.3
+      disable-cuda-graph: true
+  decode:
+    nodes: 2
+    workers: 1
+    gpus: 2  # A separate distributed decode worker on the other two nodes.
+    sidecar: true
+    kv_events: false  # Decode prefix caching is disabled by default.
+    env:
+      PYTHONUNBUFFERED: "1"
+    args:
+      served-model-name: "Qwen/Qwen3-0.6B"
+      tensor-parallel-size: 2
+      data-parallel-size: 2
+      enable-dp-attention: true
+      disaggregation-transfer-backend: nixl
+      page-size: 64
+      context-length: 2048
+      max-running-requests: 8
+      mem-fraction-static: 0.3
+      disable-cuda-graph: true
+
+benchmark:
+  type: "sa-bench"
+  isl: 128
+  osl: 64
+  concurrencies: "4"

examples/features/sglang-sidecar-multinode-dp.yaml

@@ -0,0 +1,60 @@
+# One aggregate SGLang attention-DP worker across two physical nodes.
+# Node 0 serves inference; node 1's telemetry-only sidecar relays local KV events.
+# Dynamo selects each sidecar's mode from its local engine metadata.
+# The Dynamo KV router can select either global DP rank through the same worker.
+#
+# Restart all engines and sidecars together after failure. Independent sidecar
+# restart/replay recovery is unsupported.
+
+schema: 2
+name: "qwen3-0.6b-sglang-sidecar-multinode-dp"
+
+slurm:
+  time_limit: "00:30:00"
+
+model:
+  path: "hf:Qwen/Qwen3-0.6B"
+  container: "/path/to/sglang-dynamo-multinode-sidecar.sqsh"
+  precision: "bf16"
+
+resources:
+  gpu_type: "h100"
+  gpus_per_node: 1
+
+dynamo:
+  install: false  # Dynamo is preinstalled in the image.
+  request_plane: "nats"
+  event_plane: "nats"
+
+frontend:
+  type: dynamo
+  enable_multiple_frontends: false
+  args:
+    router-mode: "kv"
+
+engine: sglang
+roles:
+  agg:
+    nodes: 2
+    workers: 1
+    gpus: 2  # Global GPU count for this one distributed worker.
+    sidecar: true
+    kv_events: true
+    env:
+      PYTHONUNBUFFERED: "1"
+    args:
+      served-model-name: "Qwen/Qwen3-0.6B"
+      tensor-parallel-size: 2
+      data-parallel-size: 2
+      enable-dp-attention: true
+      page-size: 64
+      context-length: 2048
+      max-running-requests: 8
+      mem-fraction-static: 0.3
+      disable-cuda-graph: true
+
+benchmark:
+  type: "sa-bench"
+  isl: 128
+  osl: 64
+  concurrencies: "4"

src/srtctl/backends/sglang.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 """
@@ -240,7 +240,19 @@ def endpoints_to_processes(
         from srtctl.core.topology import endpoints_to_processes, port_allocator_for
 
         allocator = port_allocator_for(port_allocator, base_sys_port)
-        processes = endpoints_to_processes(endpoints, port_allocator=allocator, sidecar_grpc=dynamo_sidecar)
+        # SGLang offsets the publisher port by global DP rank. Reserve every
+        # rank's slot so co-located workers cannot overlap a follower's range.
+        kv_events_port_sizes = {
+            endpoint.mode: _parallel_size(self.get_config_for_mode(endpoint.mode), "data", "dp")
+            for endpoint in endpoints
+            if dynamo_sidecar and self.get_kv_events_config_for_mode(endpoint.mode)
+        }
+        processes = endpoints_to_processes(
+            endpoints,
+            port_allocator=allocator,
+            sidecar_grpc=dynamo_sidecar,
+            kv_events_port_sizes=kv_events_port_sizes,
+        )
         # dist-init is bound by the endpoint's leader; every process of the
         # endpoint names the same port, allocated per leader node so two
         # endpoints led from one node do not collide.
@@ -479,11 +491,18 @@ def _build_sidecar_command(
                 "0.0.0.0",
                 "--port",
                 str(process.http_port),
-                "--nccl-port",
-                str(_nccl_port(process)),
             ]
         )
 
+        # Regular DP launches independent TP groups on this node. Let SGLang
+        # allocate their NCCL ports: an explicit port is reused by every group
+        # and causes the second DP rank to fail with EADDRINUSE.
+        regular_dp = _parallel_size(config, "data", "dp") > 1 and not (
+            config.get("enable-dp-attention") or config.get("enable_dp_attention")
+        )
+        if not regular_dp:
+            engine.extend(["--nccl-port", str(_nccl_port(process))])
+
         if mode != "agg":
             engine.extend(["--disaggregation-mode", mode, "--skip-server-warmup"])
             if mode == "prefill" and process.bootstrap_port is not None:
@@ -499,12 +518,13 @@ def _build_sidecar_command(
                     str(node_rank),
                 ]
             )
-        if is_leader:
+        kv_cfg = self.get_kv_events_config_for_mode(mode)
+        relay_kv_events = self.needs_telemetry_sidecar(process, endpoint_processes)
+        if is_leader or relay_kv_events:
             engine.extend(["--grpc-port", str(grpc_port)])
 
         # The SGLang sidecar discovers the publisher; SGLang still needs this
         # flag to enable it and advertise the topology-assigned endpoint.
-        kv_cfg = self.get_kv_events_config_for_mode(mode)
         if kv_cfg and process.kv_events_port is not None:
             kv_cfg["endpoint"] = f"tcp://*:{process.kv_events_port}"
             engine.extend(["--kv-events-config", json.dumps(kv_cfg)])
@@ -521,7 +541,7 @@ def _build_sidecar_command(
                 process.endpoint_index,
             )
         engine.extend(_config_to_cli_args(config))
-        if not is_leader:
+        if not is_leader and not relay_kv_events:
             return engine
 
         sidecar = (
@@ -530,7 +550,8 @@ def _build_sidecar_command(
             else ["python3", "-m", "dynamo.sglang.sidecar"]
         )
         sidecar.extend(["--grpc-endpoint", f"127.0.0.1:{grpc_port}"])
-        if mode == "prefill":
+        # Dynamo selects serving or telemetry mode from the local engine metadata.
+        if is_leader and mode == "prefill":
             sidecar.extend(["--bootstrap-host", leader_ip])
         sidecar.extend(sidecar_config.sidecar_args)
 
@@ -542,6 +563,53 @@ def _build_sidecar_command(
             startup_timeout=sidecar_config.sidecar_startup_timeout,
         )
 
+    def needs_telemetry_sidecar(self, process: "Process", endpoint_processes: list["Process"]) -> bool:
+        """Whether a sidecar-enabled endpoint needs a KV relay on this follower."""
+        nodes = list(dict.fromkeys(candidate.node for candidate in endpoint_processes))
+        node_rank = nodes.index(process.node)
+        kv_cfg = self.get_kv_events_config_for_mode(process.endpoint_mode)
+        return (
+            node_rank != 0
+            and kv_cfg is not None
+            and kv_cfg.get("publisher") == "zmq"
+            and process.kv_events_port is not None
+            and _node_has_kv_publisher(self.get_config_for_mode(process.endpoint_mode), len(nodes), node_rank)
+        )
+
+
+def _parallel_size(config: dict[str, Any], long_name: str, short_name: str) -> int:
+    """Read SGLang's long/short parallel-size aliases, including YAML underscores."""
+    aliases = (f"{long_name}-parallel-size", f"{short_name}-size")
+    # Match _config_to_cli_args' order when more than one alias is supplied.
+    value = 1
+    for key, configured in sorted(config.items()):
+        if key.replace("_", "-") in aliases and configured is not None:
+            value = int(configured)
+    if value < 1:
+        raise ValueError(f"SGLang {long_name}-parallel-size must be positive")
+    return value
+
+
+def _node_has_kv_publisher(config: dict[str, Any], nnodes: int, node_rank: int) -> bool:
+    """Select the nodes owning PP=0, attention CP=0/TP=0 scheduler ranks.
+
+    Attention DP partitions global TP into contiguous groups; only the first
+    rank of each group publishes. TP-only and later PP-stage followers must
+    stay engine-only because the telemetry sidecar requires a local source.
+    """
+    if not (config.get("enable-dp-attention") or config.get("enable_dp_attention")):
+        return node_rank == 0
+    tp = _parallel_size(config, "tensor", "tp")
+    dp = _parallel_size(config, "data", "dp")
+    pp = _parallel_size(config, "pipeline", "pp")
+    nodes_per_pp = max(nnodes // pp, 1)
+    if node_rank >= nodes_per_pp:
+        return False
+    if tp % nodes_per_pp or tp % dp:
+        raise ValueError("SGLang attention DP requires TP divisible by DP and nodes per PP stage")
+    local_tp = tp // nodes_per_pp
+    return any(node_rank * local_tp <= rank < (node_rank + 1) * local_tp for rank in range(0, tp, tp // dp))
+
 
 def _config_to_cli_args(config: dict[str, Any]) -> list[str]:
     """Convert config dict to CLI arguments."""

src/srtctl/core/topology.py

@@ -571,6 +571,7 @@ def endpoints_to_processes(
     port_allocator: NodePortAllocator | None = None,
     engines_per_process: int = 1,
     sidecar_grpc: bool = False,
+    kv_events_port_sizes: dict[WorkerMode, int] | None = None,
     bootstrap_ports: bool = True,
 ) -> list[Process]:
     """Convert endpoints to physical processes, one per node of each endpoint.
@@ -590,6 +591,7 @@ def endpoints_to_processes(
             engine of a node then gets its own Process (same GPUs and node_rank,
             distinct ports, ``engine_id`` 0..n-1), emitted engine 0 first.
         sidecar_grpc: Allocate a Dynamo sidecar gRPC port for every process.
+        kv_events_port_sizes: KV publisher port range per process, keyed by worker mode.
         bootstrap_ports: Allocate the prefill bootstrap port; ``False`` for an engine
             that has no bootstrap rendezvous (vLLM hands KV over its NIXL side channel).
 
@@ -626,7 +628,9 @@ def endpoints_to_processes(
                         node_rank=node_rank,
                         bootstrap_port=endpoint_bootstrap_ports[engine_id],
                         # Every process publishes KV events and opens a NIXL side channel of its own.
-                        kv_events_port=allocator.next(KV_EVENTS_PORTS),
+                        kv_events_port=allocator.next(
+                            KV_EVENTS_PORTS, size=(kv_events_port_sizes or {}).get(endpoint.mode, 1)
+                        ),
                         nixl_port=allocator.next(NIXL_PORTS),
                         het_group=endpoint.het_group,
                         engine_id=engine_id,