diff --git a/conf/experimental/ai_dynamo/test/sglang.toml b/conf/experimental/ai_dynamo/test/sglang.toml
index 8bdefb688..f4459bbf9 100644
--- a/conf/experimental/ai_dynamo/test/sglang.toml
+++ b/conf/experimental/ai_dynamo/test/sglang.toml
@@ -21,7 +21,7 @@ extra_container_mounts = ["/run/udev:/run/udev"]
 dse_excluded_args = ["cmd_args.aiperf_phases"]
 
 [cmd_args]
-docker_image_url = "nvcr.io/nvidia/ai-dynamo/sglang-runtime:1.1.1"
+docker_image_url = "nvcr.io/nvidia/ai-dynamo/sglang-runtime:1.3.1"
 workloads = "aiperf.sh"
 
   [cmd_args.dynamo]
@@ -40,7 +40,6 @@ workloads = "aiperf.sh"
       page-size = 16
       tensor-parallel-size = 1
       pipeline-parallel-size = 1
-      disaggregation-mode = "prefill"
       disaggregation-bootstrap-port = 12345
       host = "0.0.0.0"
       port = 40000
@@ -56,7 +55,6 @@ workloads = "aiperf.sh"
       page-size = 16
       tensor-parallel-size = 1
       pipeline-parallel-size = 1
-      disaggregation-mode = "decode"
       disaggregation-bootstrap-port = 12345
       host = "0.0.0.0"
       disaggregation-transfer-backend = "nixl"
@@ -127,7 +125,6 @@ UCX_LOG_LEVEL = "warn"
 HF_HUB_OFFLINE = "0"
 TRANSFORMERS_OFFLINE = "0"
 HF_DATASETS_OFFLINE = "0"
-DYNAMO_NODELIST = "$(scontrol show hostname $SLURM_JOB_NODELIST | tr -s '\\n' ',')"
 UCX_TLS = "all"
 #DYN_LOGGING_JSONL="true"
 #OTEL_EXPORT_ENABLED="1"
diff --git a/conf/experimental/ai_dynamo/test/vllm.toml b/conf/experimental/ai_dynamo/test/vllm.toml
index 20890b65d..29dc0bde4 100644
--- a/conf/experimental/ai_dynamo/test/vllm.toml
+++ b/conf/experimental/ai_dynamo/test/vllm.toml
@@ -21,7 +21,7 @@ extra_container_mounts = ["/run/udev:/run/udev"]
 dse_excluded_args = ["cmd_args.aiperf_phases"]
 
 [cmd_args]
-docker_image_url = "nvcr.io/nvidia/ai-dynamo/vllm-runtime:1.1.1"
+docker_image_url = "nvcr.io/nvidia/ai-dynamo/vllm-runtime:1.3.1"
 workloads = "aiperf.sh"
 
   [cmd_args.dynamo]
@@ -31,7 +31,7 @@ workloads = "aiperf.sh"
 
     [cmd_args.dynamo.prefill_worker]
     num-nodes = 1
-    cmd = 'python3 -m dynamo.vllm --is-prefill-worker'
+    cmd = 'python3 -m dynamo.vllm'
     worker-initialized-regex = 'VllmWorker.*has.been.initialized'
     extra-args = "--no-enable-expert-parallel"
 
@@ -75,6 +75,7 @@ workloads = "aiperf.sh"
     concurrency = 2
 
   [cmd_args.aiperf]
+  setup-cmd = "python -m pip install --break-system-packages --ignore-installed blinker==1.9.0 && python -m pip install --break-system-packages --upgrade aiperf==0.8.0"
   health-check-between-phases = true
   continue-on-phase-failure = false
     [cmd_args.aiperf.args]
@@ -125,5 +126,4 @@ UCX_LOG_LEVEL = "warn"
 HF_HUB_OFFLINE = "0"
 TRANSFORMERS_OFFLINE = "0"
 HF_DATASETS_OFFLINE = "0"
-DYNAMO_NODELIST = "$(scontrol show hostname $SLURM_JOB_NODELIST | tr -s '\\n' ',')"
 UCX_TLS = "all"
diff --git a/conf/experimental/ai_dynamo/test_scenario/sglang_hicache.toml b/conf/experimental/ai_dynamo/test_scenario/sglang_hicache.toml
index fd9fc4e96..ccde1ea40 100644
--- a/conf/experimental/ai_dynamo/test_scenario/sglang_hicache.toml
+++ b/conf/experimental/ai_dynamo/test_scenario/sglang_hicache.toml
@@ -28,7 +28,7 @@ extra_container_mounts = ["/run/udev:/run/udev"]
 dse_excluded_args = ["cmd_args.aiperf_phases"]
 
   [Tests.cmd_args]
-  docker_image_url = "nvcr.io/nvidia/ai-dynamo/sglang-runtime:1.1.1"
+  docker_image_url = "nvcr.io/nvidia/ai-dynamo/sglang-runtime:1.3.1"
   workloads = "aiperf.sh"
 
     [Tests.cmd_args.hicache.plugin.posix]
@@ -52,7 +52,6 @@ dse_excluded_args = ["cmd_args.aiperf_phases"]
         page-size = 64
         tensor-parallel-size = 1
         pipeline-parallel-size = 1
-        disaggregation-mode = "prefill"
         disaggregation-bootstrap-port = 12345
         disaggregation-transfer-backend = "nixl"
         host = "0.0.0.0"
@@ -74,7 +73,6 @@ dse_excluded_args = ["cmd_args.aiperf_phases"]
         page-size = 64
         tensor-parallel-size = 1
         pipeline-parallel-size = 1
-        disaggregation-mode = "decode"
         disaggregation-bootstrap-port = 12345
         disaggregation-transfer-backend = "nixl"
         host = "0.0.0.0"
diff --git a/conf/experimental/ai_dynamo/test_scenario/sglang_multinode_worker_slurm.toml b/conf/experimental/ai_dynamo/test_scenario/sglang_multinode_worker_slurm.toml
new file mode 100644
index 000000000..385d965a6
--- /dev/null
+++ b/conf/experimental/ai_dynamo/test_scenario/sglang_multinode_worker_slurm.toml
@@ -0,0 +1,48 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name = "dynamo-sglang-multinode-worker-slurm"
+job_status_check = false
+
+[[Tests]]
+id = "test.disagg.multinode-worker"
+test_name = "sglang"
+time_limit = "00:20:00"
+
+  [Tests.cmd_args]
+    [Tests.cmd_args.dynamo]
+    model = "Qwen/Qwen3-0.6B"
+    node-setup-cmd = "hostname"
+
+      # Work around the known SGLang multi-node TP overlap-scheduler hang:
+      # https://github.com/sgl-project/sglang/issues/26454
+      [Tests.cmd_args.dynamo.prefill_worker]
+      num-nodes = 2
+      nodes-per-worker = 2
+      extra-args = "--trust-remote-code --skip-tokenizer-init --enable-metrics --disable-overlap-schedule"
+
+        [Tests.cmd_args.dynamo.prefill_worker.args]
+        tensor-parallel-size = 8
+
+      # Work around the known SGLang multi-node TP CUDA graph replay hang:
+      # https://github.com/sgl-project/sglang/issues/29548
+      [Tests.cmd_args.dynamo.decode_worker]
+      num-nodes = 2
+      nodes-per-worker = 2
+      extra-args = "--trust-remote-code --skip-tokenizer-init --enable-metrics --disable-overlap-schedule --disable-cuda-graph"
+
+        [Tests.cmd_args.dynamo.decode_worker.args]
+        tensor-parallel-size = 8
diff --git a/conf/experimental/ai_dynamo/test_scenario/sglang_wideep_slurm.toml b/conf/experimental/ai_dynamo/test_scenario/sglang_wideep_slurm.toml
new file mode 100644
index 000000000..daca8a875
--- /dev/null
+++ b/conf/experimental/ai_dynamo/test_scenario/sglang_wideep_slurm.toml
@@ -0,0 +1,68 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name = "dynamo-sglang-qwen3-30b-a3b-fp8-wideep-slurm"
+job_status_check = false
+
+[[Tests]]
+id = "test.disagg.1p1d.multinode-wideep"
+test_name = "sglang"
+time_limit = "00:30:00"
+
+  [Tests.cmd_args]
+    [Tests.cmd_args.dynamo]
+    model = "Qwen/Qwen3-30B-A3B-FP8"
+    node-setup-cmd = "hostname"
+
+      # Work around the SGLang multi-node overlap-scheduler hang:
+      # https://github.com/sgl-project/sglang/issues/26454
+      [Tests.cmd_args.dynamo.prefill_worker]
+      num-nodes = 2
+      nodes-per-worker = 2
+      extra-args = "--trust-remote-code --skip-tokenizer-init --enable-metrics --enable-dp-attention --disable-overlap-schedule"
+
+        [Tests.cmd_args.dynamo.prefill_worker.args]
+        model-path = "Qwen/Qwen3-30B-A3B-FP8"
+        served-model-name = "Qwen/Qwen3-30B-A3B-FP8"
+        tensor-parallel-size = 8
+        pipeline-parallel-size = 1
+        data-parallel-size = 8
+        ep-size = 8
+        moe-a2a-backend = "deepep"
+        deepep-mode = "normal"
+        mem-fraction-static = 0.75
+        load-balance-method = "round_robin"
+        watchdog-timeout = 3600
+
+      # Work around the SGLang multi-node overlap-scheduler and CUDA graph replay hangs:
+      # https://github.com/sgl-project/sglang/issues/26454
+      # https://github.com/sgl-project/sglang/issues/29548
+      [Tests.cmd_args.dynamo.decode_worker]
+      num-nodes = 2
+      nodes-per-worker = 2
+      extra-args = "--trust-remote-code --skip-tokenizer-init --enable-metrics --enable-dp-attention --prefill-round-robin-balance --disable-overlap-schedule --disable-cuda-graph"
+
+        [Tests.cmd_args.dynamo.decode_worker.args]
+        model-path = "Qwen/Qwen3-30B-A3B-FP8"
+        served-model-name = "Qwen/Qwen3-30B-A3B-FP8"
+        tensor-parallel-size = 8
+        pipeline-parallel-size = 1
+        data-parallel-size = 8
+        ep-size = 8
+        moe-a2a-backend = "deepep"
+        deepep-mode = "normal"
+        mem-fraction-static = 0.75
+        watchdog-timeout = 3600
diff --git a/conf/experimental/ai_dynamo/test_scenario/vllm_lmcache.toml b/conf/experimental/ai_dynamo/test_scenario/vllm_lmcache.toml
index 09672c786..030d21814 100644
--- a/conf/experimental/ai_dynamo/test_scenario/vllm_lmcache.toml
+++ b/conf/experimental/ai_dynamo/test_scenario/vllm_lmcache.toml
@@ -30,7 +30,7 @@ dse_excluded_args = [
 ]
 
   [Tests.cmd_args]
-  docker_image_url = "nvcr.io/nvidia/ai-dynamo/vllm-runtime:1.1.1"
+  docker_image_url = "nvcr.io/nvidia/ai-dynamo/vllm-runtime:1.3.1"
   # storage_cache_dir = "/lustre/.../install/tmp"
   workloads = "aiperf.sh"
 
@@ -61,7 +61,7 @@ dse_excluded_args = [
 
       [Tests.cmd_args.dynamo.prefill_worker]
       num-nodes = 1
-      cmd = 'python3 -m dynamo.vllm --is-prefill-worker'
+      cmd = 'python3 -m dynamo.vllm'
       worker-initialized-regex = 'VllmWorker.*has.been.initialized'
       extra-args = "--no-enable-expert-parallel"
 
diff --git a/conf/experimental/ai_dynamo/test_scenario/vllm_multinode_worker_slurm.toml b/conf/experimental/ai_dynamo/test_scenario/vllm_multinode_worker_slurm.toml
new file mode 100644
index 000000000..bda6ab3af
--- /dev/null
+++ b/conf/experimental/ai_dynamo/test_scenario/vllm_multinode_worker_slurm.toml
@@ -0,0 +1,54 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name = "dynamo-vllm-multinode-worker-slurm"
+job_status_check = false
+
+[[Tests]]
+id = "test.disagg.multinode-worker"
+test_name = "vLLM"
+time_limit = "00:10:00"
+
+  [Tests.cmd_args]
+    [Tests.cmd_args.dynamo.dcgm_exporter]
+    enabled = true
+    docker-image-url = "nvcr.io/nvidia/k8s/dcgm-exporter:4.5.2-4.8.1-distroless"
+
+    [Tests.cmd_args.dynamo.prefill_worker]
+    num-nodes = 2
+    nodes-per-worker = 2
+      [Tests.cmd_args.dynamo.prefill_worker.args]
+      pipeline-parallel-size = 1
+
+    [Tests.cmd_args.dynamo.decode_worker]
+    num-nodes = 2
+    nodes-per-worker = 2
+      [Tests.cmd_args.dynamo.decode_worker.args]
+      pipeline-parallel-size = 1
+
+    [[Tests.cmd_args.aiperf_phases]]
+    name = "round_1"
+      [Tests.cmd_args.aiperf_phases.args]
+      concurrency = 4
+      request-count = 50
+      server-metrics = "auto"
+
+    [[Tests.cmd_args.aiperf_phases]]
+    name = "round_2"
+      [Tests.cmd_args.aiperf_phases.args]
+      concurrency = 8
+      request-count = 50
+      server-metrics = "auto"
diff --git a/conf/experimental/ai_dynamo/test_scenario/vllm_wideep_slurm.toml b/conf/experimental/ai_dynamo/test_scenario/vllm_wideep_slurm.toml
new file mode 100644
index 000000000..4a2b60fec
--- /dev/null
+++ b/conf/experimental/ai_dynamo/test_scenario/vllm_wideep_slurm.toml
@@ -0,0 +1,55 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name = "dynamo-vllm-qwen3-30b-a3b-wideep-slurm"
+job_status_check = false
+
+[[Tests]]
+id = "test.disagg.1p1d.multinode-wideep"
+test_name = "vLLM"
+time_limit = "00:30:00"
+
+  [Tests.cmd_args]
+    [Tests.cmd_args.dynamo]
+    model = "Qwen/Qwen3-30B-A3B"
+    node-setup-cmd = "hostname"
+
+      # On four-GPU nodes, TP=1 and DP=8 place four DP ranks per node.
+      # vLLM derives EP=TP*DP=8, so each role's expert group spans both nodes.
+      [Tests.cmd_args.dynamo.prefill_worker]
+      num-nodes = 2
+      nodes-per-worker = 2
+      extra-args = "--enable-expert-parallel"
+
+        [Tests.cmd_args.dynamo.prefill_worker.args]
+        tensor-parallel-size = 1
+        pipeline-parallel-size = 1
+        data-parallel-size = 8
+        all2all-backend = "allgather_reducescatter"
+        max-model-len = 8192
+        kv-events-config = '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20080","enable_kv_cache_events":true}'
+
+      [Tests.cmd_args.dynamo.decode_worker]
+      num-nodes = 2
+      nodes-per-worker = 2
+      extra-args = "--enable-expert-parallel"
+
+        [Tests.cmd_args.dynamo.decode_worker.args]
+        tensor-parallel-size = 1
+        pipeline-parallel-size = 1
+        data-parallel-size = 8
+        all2all-backend = "allgather_reducescatter"
+        max-model-len = 8192
diff --git a/doc/workloads/ai_dynamo.rst b/doc/workloads/ai_dynamo.rst
index c024430f9..f3e474d23 100644
--- a/doc/workloads/ai_dynamo.rst
+++ b/doc/workloads/ai_dynamo.rst
@@ -63,6 +63,63 @@ run on the same allocated node(s) with separate GPU slices.
 
 All node role assignments and orchestration are automatically managed by CloudAI.
 
+Multinode Backend Workers on Slurm
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+``prefill_worker.num-nodes`` and ``decode_worker.num-nodes`` specify the total nodes assigned to each role.
+Set ``nodes-per-worker`` to group those nodes into logical backend workers. For example, the following configuration
+creates one prefill worker spanning two nodes and one decode worker spanning two other nodes:
+
+.. code-block:: toml
+
+   [cmd_args.dynamo.prefill_worker]
+   num-nodes = 2
+   nodes-per-worker = 2
+     [cmd_args.dynamo.prefill_worker.args]
+     tensor-parallel-size = 8
+     pipeline-parallel-size = 1
+
+   [cmd_args.dynamo.decode_worker]
+   num-nodes = 2
+   nodes-per-worker = 2
+     [cmd_args.dynamo.decode_worker.args]
+     tensor-parallel-size = 8
+     pipeline-parallel-size = 1
+
+When ``nodes-per-worker`` is omitted, it defaults to ``1`` and preserves the previous one-worker-per-node behavior.
+For vLLM TP/PP groups, CloudAI launches rank 0 as the Dynamo-facing process and the remaining ranks with
+``--headless`` using Dynamo's multiprocessing executor. When ``data-parallel-size`` is greater than one, CloudAI
+instead launches a full Dynamo vLLM process on every node and derives ``data-parallel-size-local``,
+``data-parallel-start-rank``, the coordinator address, and the RPC port from the worker topology. SGLang receives
+``--dist-init-addr``, ``--nnodes``, and ``--node-rank`` on every participating node; its configured
+``data-parallel-size`` is forwarded as ``dp-size``. The launcher derives prefill and decode disaggregation modes
+from the configured roles, so worker commands do not need role flags.
+
+For TP/PP-spanning workers, ``tensor-parallel-size * pipeline-parallel-size`` must be divisible by
+``nodes-per-worker``. For vLLM multinode data parallelism, ``data-parallel-size`` must instead be divisible by
+``nodes-per-worker``, and each DP rank's TP/PP group must fit within one node. SGLang multinode data parallelism
+requires ``--enable-dp-attention``. The resulting per-node GPU footprint must fit on each node. Ray-backed Dynamo
+vLLM groups are not currently supported. The Slurm launcher explicitly selects etcd discovery and the NATS event
+plane.
+
+No separate WideEP mode is required. Multinode data-parallel orchestration is inferred from
+``nodes-per-worker > 1`` and ``data-parallel-size > 1``. Backend expert-parallel parameters such as vLLM's
+``--enable-expert-parallel`` or SGLang's ``ep-size`` and ``--enable-dp-attention`` remain explicit backend
+configuration.
+CloudAI uses the allocated Slurm node names for multinode rendezvous. The cluster must make those names consistently
+resolvable and mutually reachable from every allocated node and workload container. Multi-homed systems should also
+configure backend network selection, such as ``GLOO_SOCKET_IFNAME`` and ``NCCL_SOCKET_IFNAME``, at the system level.
+Explicit backend rendezvous addresses take precedence over addresses derived from the Slurm node names.
+Clusters that cannot provide this hostname contract must supply backend-native per-node address overrides, such as
+``SGLANG_HOST_IP`` or ``VLLM_NIXL_SIDE_CHANNEL_HOST``, together with explicit worker rendezvous arguments.
+
+Dedicated examples are available in ``test_scenario/vllm_multinode_worker_slurm.toml`` and
+``test_scenario/sglang_multinode_worker_slurm.toml``. The existing ``vllm_slurm.toml`` and ``sglang_slurm.toml``
+scenarios retain their previous configurations. The vLLM example inherits ``tensor-parallel-size = 8`` from
+``test/vllm.toml``; the SGLang example sets it explicitly because ``test/sglang.toml`` defaults to ``1``.
+WideEP examples are provided in ``test_scenario/vllm_wideep_slurm.toml`` and
+``test_scenario/sglang_wideep_slurm.toml``.
+
 Launch and Monitor the Job
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -405,8 +462,8 @@ Shared-Node Disaggregated Runs
 For Slurm, set top-level ``num_nodes`` lower than the sum of ``prefill_worker.num-nodes`` and
 ``decode_worker.num-nodes`` to run both roles on the same allocated node(s). For example, ``num_nodes = 1`` with
 ``prefill_worker.num-nodes = 1`` and ``decode_worker.num-nodes = 1`` runs one prefill worker and one decode worker on
-the same node. CloudAI assigns decode GPUs first and prefill GPUs after that based on each role's
-``tensor-parallel-size * pipeline-parallel-size``. The combined role GPU count must fit on one node.
+the same node. CloudAI assigns decode GPUs first and prefill GPUs after that based on each role's per-node GPU
+footprint, including local data-parallel ranks where applicable. The combined role GPU count must fit on one node.
 
 Example ``aiperf_report.csv``:
 
@@ -432,8 +489,12 @@ Supported Backends
 The following backends are available via the ``conf/experimental/ai_dynamo/test/`` directory:
 
 - **vLLM** (``vllm.toml``) — use with ``test_scenario/vllm_slurm.toml``
+- **vLLM multinode worker** — use ``test_scenario/vllm_multinode_worker_slurm.toml``
+- **vLLM WideEP** — use ``test_scenario/vllm_wideep_slurm.toml``
 - **vLLM with LMCache config propagation** — use self-contained scenario ``test_scenario/vllm_lmcache.toml``
 - **sglang** (``sglang.toml``) — use with ``test_scenario/sglang_slurm.toml``
+- **SGLang multinode worker** — use ``test_scenario/sglang_multinode_worker_slurm.toml``
+- **SGLang WideEP** — use ``test_scenario/sglang_wideep_slurm.toml``
 
 Both backends use ``aiperf`` as the default benchmark tool and support disaggregated prefill/decode.
 
diff --git a/src/cloudai/workloads/ai_dynamo/ai_dynamo.py b/src/cloudai/workloads/ai_dynamo/ai_dynamo.py
index 984902bcd..176f803c3 100644
--- a/src/cloudai/workloads/ai_dynamo/ai_dynamo.py
+++ b/src/cloudai/workloads/ai_dynamo/ai_dynamo.py
@@ -16,6 +16,7 @@
 
 import csv
 import logging
+import shlex
 from pathlib import Path
 from typing import Literal, Optional, cast
 
@@ -47,6 +48,7 @@
 LMCACHE_CONFIG_FILE_NAME = "lmcache-config.yaml"
 LMCACHE_CONFIG_BACKUP_FILE_NAME = "lmcache-config.original.yaml"
 HICACHE_CONFIG_FILE_NAME = "hicache-config.toml"
+DYNAMO_COMMIT = "a49702e4432e7fa43cbc88175bddb31604340f19"
 
 
 class Args(BaseModel):
@@ -109,6 +111,11 @@ class WorkerBaseArgs(Args):
         serialization_alias="data-parallel-size",
         validation_alias=AliasChoices("data-parallel-size", "data_parallel_size"),
     )
+    distributed_executor_backend: Literal["mp", "ray"] | None = Field(
+        default=None,
+        serialization_alias="distributed-executor-backend",
+        validation_alias=AliasChoices("distributed-executor-backend", "distributed_executor_backend"),
+    )
 
 
 class WorkerConfig(BaseModel):
@@ -131,6 +138,15 @@ class WorkerConfig(BaseModel):
     num_nodes: int | list[int] = Field(
         default=1, serialization_alias="num-nodes", validation_alias=AliasChoices("num-nodes", "num_nodes")
     )
+    nodes_per_worker: int | list[int] | None = Field(
+        default=None,
+        description=(
+            "Number of physical nodes in one logical Slurm backend worker. When omitted, it defaults to one node "
+            "per worker and preserves the legacy Slurm behavior."
+        ),
+        serialization_alias="nodes-per-worker",
+        validation_alias=AliasChoices("nodes-per-worker", "nodes_per_worker"),
+    )
     nodes: str | None = Field(default=None)
 
     args: WorkerBaseArgs = Field(default_factory=WorkerBaseArgs)
@@ -160,7 +176,32 @@ def validate_enabled_worker_fields(self) -> "WorkerConfig":
             missing_fields.append("worker-initialized-regex")
         if missing_fields:
             raise ValueError(f"{', '.join(missing_fields)} must be set when num-nodes is non-zero")
+        return self
+
+    def has_extra_arg(self, option: str) -> bool:
+        """Return whether an option is present in the worker's backend arguments."""
+        values = [self.extra_args] if isinstance(self.extra_args, str) else self.extra_args or []
+        return any(
+            token == option or token.startswith(f"{option}=") for value in values for token in shlex.split(value)
+        )
 
+    @model_validator(mode="after")
+    def validate_worker_topology(self) -> "WorkerConfig":
+        """Validate scalar worker topology while allowing DSE lists before unrolling."""
+        if isinstance(self.num_nodes, list) or isinstance(self.nodes_per_worker, list):
+            return self
+
+        nodes_per_worker = self.nodes_per_worker
+        if nodes_per_worker is None:
+            return self
+        if nodes_per_worker < 1:
+            raise ValueError("nodes_per_worker must be at least 1")
+        if self.num_nodes < 1:
+            raise ValueError("num_nodes must be at least 1 when nodes_per_worker is set")
+        if self.num_nodes % nodes_per_worker != 0:
+            raise ValueError("num_nodes must be divisible by nodes_per_worker")
+        if nodes_per_worker > 1 and self.multiple_workers_per_node:
+            raise ValueError("multiple_workers_per_node is incompatible with nodes_per_worker > 1")
         return self
 
 
@@ -250,7 +291,7 @@ def validate_connector(cls, v: str | list[str] | None) -> str | list[str] | None
         worker_initialized_regex="VllmWorker.*has.been.initialized",
     )
     prefill_worker: WorkerConfig = WorkerConfig(
-        cmd="python3 -m dynamo.vllm --is-prefill-worker",
+        cmd="python3 -m dynamo.vllm",
         worker_initialized_regex="VllmWorker.*has.been.initialized",
     )
 
@@ -418,7 +459,7 @@ class AIDynamoCmdArgs(CmdArgs):
     model_config = ConfigDict(extra="forbid")
 
     dynamo_version: str = Field(
-        default="f7e468c7e8ff0d1426db987564e60572167e8464",
+        default=DYNAMO_COMMIT,
         description="AI Dynamo Git commit, tag, or branch.",
     )
     docker_image_url: str
@@ -594,7 +635,16 @@ def _was_workload_successful(self, output_path: Path, workload: str, workload_ma
             logging.info(f"Workload {workload} not found in workload map")
             return False
 
-        return self._was_workload_report_produced(output_path, workload, workload_config)
+        if not self._was_workload_report_produced(output_path, workload, workload_config):
+            return False
+
+        if isinstance(workload_config, AIPerf):
+            request_count = parse_aiperf_request_count(output_path / workload_config.report_name)
+            if request_count is None or request_count <= 0:
+                logging.info(f"AIPerf report has no successful requests: {output_path / workload_config.report_name}")
+                return False
+
+        return True
 
     def _were_workloads_successful(self, output_path: Path) -> bool:
         workload_map = self.get_workload_map()
@@ -629,29 +679,69 @@ def constraint_check(self, tr: TestRun, system: Optional[System]) -> bool:
         prefill_worker = tr.test.cmd_args.dynamo.prefill_worker
         decode_worker = tr.test.cmd_args.dynamo.decode_worker
 
-        prefill_tp = prefill_worker.args.tensor_parallel_size
-        prefill_pp = prefill_worker.args.pipeline_parallel_size
-
-        decode_tp = decode_worker.args.tensor_parallel_size
-        decode_pp = decode_worker.args.pipeline_parallel_size
+        prefill_tp = int(prefill_worker.args.tensor_parallel_size)
+        prefill_pp = int(prefill_worker.args.pipeline_parallel_size)
+        decode_tp = int(decode_worker.args.tensor_parallel_size)
+        decode_pp = int(decode_worker.args.pipeline_parallel_size)
 
         if self.constraints.prefill_tp_le_decode_tp and prefill_tp > decode_tp:
             logging.info("constraint_check failed for: prefill_tp_le_decode_tp")
             return False
         logging.info("constraint_check passed for: prefill_tp_le_decode_tp")
 
-        gpus_per_node = 0
-        slurm_system = cast(SlurmSystem, system)
-        if slurm_system and slurm_system.gpus_per_node:
-            gpus_per_node = slurm_system.gpus_per_node
-
-        if (
-            gpus_per_node > 0
-            and self.constraints.tp_times_pp_le_gpus_per_node
-            and (prefill_tp * prefill_pp > gpus_per_node or decode_tp * decode_pp > gpus_per_node)
+        gpus_per_node = int(getattr(cast(SlurmSystem, system), "gpus_per_node", 0) or 0)
+        role_footprints: dict[str, int] = {}
+        for role, worker, tp, pp in (
+            ("prefill", prefill_worker, prefill_tp, prefill_pp),
+            ("decode", decode_worker, decode_tp, decode_pp),
         ):
-            logging.info("constraint_check failed for: tp_times_pp_le_gpus_per_node")
-            return False
+            num_nodes = int(worker.num_nodes)
+            nodes_per_worker = int(worker.nodes_per_worker or 1)
+            world_size = tp * pp
+            data_parallel_size = int(worker.args.data_parallel_size or 1)
+            backend = tr.test.cmd_args.dynamo.backend
+            is_vllm_multinode_dp = backend == "vllm" and nodes_per_worker > 1 and data_parallel_size > 1
+
+            if num_nodes == 0 and worker.nodes_per_worker is None:
+                role_footprints[role] = 0
+                continue
+            if (
+                backend == "sglang"
+                and nodes_per_worker > 1
+                and data_parallel_size > 1
+                and not worker.has_extra_arg("--enable-dp-attention")
+            ):
+                logging.info("constraint_check failed: multinode SGLang DP requires --enable-dp-attention")
+                return False
+            if (
+                backend == "vllm"
+                and nodes_per_worker > 1
+                and worker.args.distributed_executor_backend not in {None, "mp"}
+            ):
+                logging.info("constraint_check failed: multinode vLLM requires the mp executor")
+                return False
+            if (
+                nodes_per_worker < 1
+                or num_nodes < 1
+                or num_nodes % nodes_per_worker != 0
+                or world_size < 1
+                or (is_vllm_multinode_dp and data_parallel_size % nodes_per_worker != 0)
+                or (not is_vllm_multinode_dp and world_size % nodes_per_worker != 0)
+                or (nodes_per_worker > 1 and worker.multiple_workers_per_node)
+            ):
+                logging.info("constraint_check failed for invalid %s worker topology", role)
+                return False
+
+            local_footprint = (
+                world_size * (data_parallel_size // nodes_per_worker)
+                if is_vllm_multinode_dp
+                else world_size // nodes_per_worker
+            )
+            if gpus_per_node > 0 and self.constraints.tp_times_pp_le_gpus_per_node and local_footprint > gpus_per_node:
+                logging.info("constraint_check failed for %s worker GPU capacity", role)
+                return False
+            role_footprints[role] = local_footprint
+
         logging.info("constraint_check passed for: tp_times_pp_le_gpus_per_node")
 
         role_total_nodes = int(prefill_worker.num_nodes) + int(decode_worker.num_nodes)
@@ -665,7 +755,7 @@ def constraint_check(self, tr: TestRun, system: Optional[System]) -> bool:
             shared_node_disagg
             and gpus_per_node > 0
             and self.constraints.tp_times_pp_le_gpus_per_node
-            and (prefill_tp * prefill_pp + decode_tp * decode_pp > gpus_per_node)
+            and (role_footprints["prefill"] + role_footprints["decode"] > gpus_per_node)
         ):
             logging.info("constraint_check failed for: shared_node_tp_pp_sum_le_gpus_per_node")
             return False
@@ -708,6 +798,21 @@ def _parse_count_value(value: str | int | float | None) -> float | None:
         return None
 
 
+def parse_aiperf_request_count(report_path: Path) -> float | None:
+    """Return the number of successful requests recorded by AIPerf."""
+    if not report_path.exists():
+        return None
+
+    try:
+        with report_path.open(newline="", encoding="utf-8") as csv_file:
+            for row in csv.reader(csv_file):
+                if len(row) >= 2 and row[0].strip() == "Request Count":
+                    return _parse_count_value(row[1])
+    except (OSError, csv.Error):
+        return None
+    return None
+
+
 def parse_aiperf_accuracy(output_path: Path) -> float | None:
     """
     Parse AIPerf accuracy from accuracy_results.csv.
diff --git a/src/cloudai/workloads/ai_dynamo/ai_dynamo.sh b/src/cloudai/workloads/ai_dynamo/ai_dynamo.sh
index f0612b2db..35877172a 100644
--- a/src/cloudai/workloads/ai_dynamo/ai_dynamo.sh
+++ b/src/cloudai/workloads/ai_dynamo/ai_dynamo.sh
@@ -57,7 +57,8 @@ dynamo_args["etcd-cmd"]="etcd --log-level debug"
 dynamo_args["nats-cmd"]="nats-server -js"
 dynamo_args["worker-error-pattern"]="zmq.error.ZMQError:.Address.already.in.use|ERROR.core.run_engine_core:.EngineCore.failed.to.start|ERROR.multiproc_executor.worker_busy_loop:.WorkerProc.hit.an.exception|ValueError:.a.python.*async.generator:.EngineDeadError:.EngineCore.encountered.an.issue|ZeroDivisionError:.integer.division.or.modulo.by.zero|ERROR.core.run_engine_core:.EngineCore.encountered.a.fatal.error|Exception:.Failed.to.fetch.model|ERROR.*Engine.core.proc.EngineCore_.*died.unexpectedly|RuntimeError:.Engine.core.initialization.failed.|pydantic_core._pydantic_core.ValidationError|Unsupported.connector.type|error: unrecognized arguments:|tomllib.TOMLDecodeError"
 
-# sglang_dsr1-specific optional ports. Ignored by vllm.
+# Backend rendezvous ports. Prefill and decode use distinct ports so their
+# distributed groups can share physical nodes safely.
 dynamo_args["sgl-http-port"]=9001
 dynamo_args["prefill-port"]=30011
 dynamo_args["decode-port"]=30021
@@ -106,34 +107,216 @@ _gpus_per_node() {
   [[ "$n" -gt 0 ]] && echo "$n" || echo "1"
 }
 
-_resolve_host_ip() {
-  local host="$1"
-  local ip
-  ip="$(getent ahosts "$host" | grep STREAM | head -n1 | awk '{print $1}')"
-  if [[ -z "$ip" ]]; then
-    log "ERROR: Could not resolve IP for host $host"
-    exit 1
+_normalize_worker_topology() {
+  local role
+  for role in prefill decode; do
+    local -n role_config="${role}_config"
+    if [[ -z "${role_config["nodes-per-worker"]:-}" ]]; then
+      if _is_sglang_dsr1 && [[ "${role_config["num-nodes"]}" -gt 0 ]]; then
+        role_config["nodes-per-worker"]="${role_config["num-nodes"]}"
+      else
+        role_config["nodes-per-worker"]=1
+      fi
+    fi
+  done
+}
+
+_role_node_index() {
+  local role="$1"
+  local -n role_config="${role}_config"
+  _csv_index_of "${role_config["node-list"]}" "$(_current_node_name)"
+}
+
+_role_group_rank() {
+  local role="$1"
+  local -n role_config="${role}_config"
+  local node_index
+  node_index="$(_role_node_index "$role")"
+  if [[ "$node_index" -lt 0 ]]; then
+    echo "-1"
+    return
+  fi
+  echo $(( node_index % role_config["nodes-per-worker"] ))
+}
+
+_role_group_leader() {
+  local role="$1"
+  local -n role_config="${role}_config"
+  local node_index leader_index
+  local -a role_nodes
+  node_index="$(_role_node_index "$role")"
+  if [[ "$node_index" -lt 0 ]]; then
+    echo ""
+    return
+  fi
+  leader_index=$(( (node_index / role_config["nodes-per-worker"]) * role_config["nodes-per-worker"] ))
+  IFS=',' read -ra role_nodes <<< "${role_config["node-list"]}"
+  echo "${role_nodes[$leader_index]}"
+}
+
+_role_is_multinode() {
+  local role="$1"
+  local -n role_config="${role}_config"
+  [[ "${role_config["nodes-per-worker"]}" -gt 1 ]]
+}
+
+_role_data_parallel_size() {
+  local role="$1"
+  local -n role_args="${role}_args"
+  echo "${role_args["--data-parallel-size"]:-${role_args["--dp-size"]:-1}}"
+}
+
+_normalize_sglang_data_parallel_size() {
+  local role="$1"
+  local -n role_args="${role}_args"
+  if [[ -n "${role_args["--data-parallel-size"]:-}" ]]; then
+    role_args["--dp-size"]="${role_args["--data-parallel-size"]}"
+    unset 'role_args["--data-parallel-size"]'
   fi
-  echo "$ip"
 }
 
-_current_node_ip() {
-  _resolve_host_ip "$(_current_node_name)"
+_role_has_extra_arg() {
+  local role="$1"
+  local option="$2"
+  local -n role_config="${role}_config"
+  local arg
+  local -a extra_args
+  read -ra extra_args <<< "${role_config["extra-args"]:-}"
+  for arg in "${extra_args[@]}"; do
+    if [[ "$arg" == "$option" || "$arg" == "$option="* ]]; then
+      return 0
+    fi
+  done
+  return 1
+}
+
+_role_uses_vllm_multinode_dp() {
+  local role="$1"
+  _is_vllm && _role_is_multinode "$role" && [[ "$(_role_data_parallel_size "$role")" -gt 1 ]]
+}
+
+_role_is_group_leader() {
+  [[ "$(_role_group_rank "$1")" -eq 0 ]]
+}
+
+_role_is_vllm_headless() {
+  local role="$1"
+  _is_vllm && _role_is_multinode "$role" && ! _role_uses_vllm_multinode_dp "$role" \
+    && ! _role_is_group_leader "$role"
+}
+
+_validate_worker_topology() {
+  local role
+  for role in prefill decode; do
+    local -n role_config="${role}_config"
+    local -n role_args="${role}_args"
+    local num_nodes="${role_config["num-nodes"]}"
+    local nodes_per_worker="${role_config["nodes-per-worker"]}"
+    local tp="${role_args["--tensor-parallel-size"]}"
+    local pp="${role_args["--pipeline-parallel-size"]}"
+    local dp="$(_role_data_parallel_size "$role")"
+    local world_size=$(( tp * pp ))
+
+    if [[ "$num_nodes" -eq 0 ]]; then
+      continue
+    fi
+    if [[ "$nodes_per_worker" -lt 1 || "$num_nodes" -lt 1 || $((num_nodes % nodes_per_worker)) -ne 0 ]]; then
+      log "ERROR: ${role} num-nodes (${num_nodes}) must be positive and divisible by nodes-per-worker (${nodes_per_worker})"
+      exit 1
+    fi
+    # The specialized DSR1 path derives TP/DP from the allocated node count
+    # after this structural validation.
+    if _is_sglang_dsr1; then
+      continue
+    fi
+    if _is_sglang && _role_is_multinode "$role" && [[ "$dp" -gt 1 ]] \
+      && ! _role_has_extra_arg "$role" "--enable-dp-attention"; then
+      log "ERROR: ${role} multinode SGLang data parallelism requires --enable-dp-attention"
+      exit 1
+    fi
+    if _is_vllm && [[ "$dp" -gt 1 && $((dp % nodes_per_worker)) -ne 0 ]]; then
+      log "ERROR: ${role} data-parallel-size (${dp}) must be divisible by nodes-per-worker (${nodes_per_worker})"
+      exit 1
+    fi
+    if ! { _is_vllm && [[ "$dp" -gt 1 ]]; } && [[ $((world_size % nodes_per_worker)) -ne 0 ]]; then
+      log "ERROR: ${role} TP*PP (${world_size}) must be divisible by nodes-per-worker (${nodes_per_worker})"
+      exit 1
+    fi
+    if [[ "$nodes_per_worker" -gt 1 && "${role_config["multiple-workers-per-node"],,}" == "true" ]]; then
+      log "ERROR: ${role} multiple-workers-per-node is incompatible with nodes-per-worker > 1"
+      exit 1
+    fi
+  done
+}
+
+_apply_multinode_role_args() {
+  local role="$1"
+  local -n role_config="${role}_config"
+  local -n role_args="${role}_args"
+
+  if ! _role_is_multinode "$role"; then
+    return
+  fi
+
+  local node_rank leader_host rendezvous_port
+  node_rank="$(_role_group_rank "$role")"
+  if [[ "$node_rank" -lt 0 ]]; then
+    return
+  fi
+  leader_host="$(_role_group_leader "$role")"
+  rendezvous_port="${dynamo_args["${role}-port"]}"
+
+  role_config["group-rank"]="$node_rank"
+  role_config["group-leader"]="$leader_host"
+
+  if _is_vllm; then
+    if [[ -n "${role_args["--distributed-executor-backend"]:-}" && "${role_args["--distributed-executor-backend"]}" != "mp" ]]; then
+      log "ERROR: Multinode Dynamo vLLM supports only --distributed-executor-backend mp"
+      exit 1
+    fi
+    role_args["--distributed-executor-backend"]="mp"
+    if _role_uses_vllm_multinode_dp "$role"; then
+      local dp local_dp_size
+      dp="$(_role_data_parallel_size "$role")"
+      local_dp_size=$(( dp / role_config["nodes-per-worker"] ))
+      role_args["--data-parallel-size-local"]="$local_dp_size"
+      role_args["--data-parallel-start-rank"]=$(( node_rank * local_dp_size ))
+      role_args["--data-parallel-address"]="${role_args["--data-parallel-address"]:-$leader_host}"
+      role_args["--data-parallel-rpc-port"]="${role_args["--data-parallel-rpc-port"]:-$rendezvous_port}"
+      unset 'role_args["--nnodes"]'
+      unset 'role_args["--node-rank"]'
+      unset 'role_args["--master-addr"]'
+      unset 'role_args["--master-port"]'
+    else
+      role_args["--nnodes"]="${role_config["nodes-per-worker"]}"
+      role_args["--node-rank"]="$node_rank"
+      role_args["--master-addr"]="${role_args["--master-addr"]:-$leader_host}"
+      role_args["--master-port"]="${role_args["--master-port"]:-$rendezvous_port}"
+    fi
+  elif _is_sglang; then
+    _normalize_sglang_data_parallel_size "$role"
+    role_args["--dist-init-addr"]="${role_args["--dist-init-addr"]:-${leader_host}:${rendezvous_port}}"
+    role_args["--nnodes"]="${role_config["nodes-per-worker"]}"
+    role_args["--node-rank"]="$node_rank"
+  fi
 }
 
 _apply_sglang_dsr1_section_args() {
   local self="$(_current_node_name)"
   local gpn="$(_gpus_per_node)"
-  local deepep_path="${dynamo_args["repo"]}/components/backends/sglang/configs/deepseek_r1/wideep/deepep.json"
+  local deepep_path="${dynamo_args["repo"]}/recipes/deepseek-r1/sglang/deepep.json"
+  _normalize_sglang_data_parallel_size prefill
+  _normalize_sglang_data_parallel_size decode
 
   # prefill group
   local prefill_nodes="${prefill_config["num-nodes"]}"
   if [[ "$prefill_nodes" -gt 0 ]]; then
     local prefill_master_host="$(_first_in_csv "${prefill_config["node-list"]}")"
-    local prefill_master_ip="$(_resolve_host_ip "${prefill_master_host}")"
     local prefill_rank="$(_csv_index_of "${prefill_config["node-list"]}" "$self")"
     local prefill_total_gpus=$(( gpn * prefill_nodes ))
-    prefill_args["--dist-init-addr"]="${prefill_master_ip}:${dynamo_args["prefill-port"]}"
+    if [[ -z "${prefill_args["--dist-init-addr"]:-}" ]]; then
+      prefill_args["--dist-init-addr"]="${prefill_master_host}:${dynamo_args["prefill-port"]}"
+    fi
     prefill_args["--nnodes"]="${prefill_nodes}"
     prefill_args["--node-rank"]="$([[ "$prefill_rank" -ge 0 ]] && echo "$prefill_rank" || echo 0)"
     prefill_args["--tp-size"]="${prefill_args["--tp-size"]:-${prefill_total_gpus}}"
@@ -143,16 +326,17 @@ _apply_sglang_dsr1_section_args() {
   # decode group
   local decode_nodes="${decode_config["num-nodes"]}"
   local decode_master_host="$(_first_in_csv "${decode_config["node-list"]}")"
-  local decode_master_ip="$(_resolve_host_ip "${decode_master_host}")"
   local decode_rank="$(_csv_index_of "${decode_config["node-list"]}" "$self")"
   local decode_total_gpus=$(( gpn * decode_nodes ))
-  decode_args["--dist-init-addr"]="${decode_master_ip}:${dynamo_args["decode-port"]}"
+  if [[ -z "${decode_args["--dist-init-addr"]:-}" ]]; then
+    decode_args["--dist-init-addr"]="${decode_master_host}:${dynamo_args["decode-port"]}"
+  fi
   decode_args["--nnodes"]="${decode_nodes}"
   decode_args["--node-rank"]="$([[ "$decode_rank" -ge 0 ]] && echo "$decode_rank" || echo 0)"
   decode_args["--tp-size"]="${decode_args["--tp-size"]:-${decode_total_gpus}}"
   decode_args["--dp-size"]="${decode_args["--dp-size"]:-${decode_total_gpus}}"
 
-  [[ -f "$deepep_path" ]] || log "WARN: deepep-config not found: ${dynamo_args["deepep-config"]}"
+  [[ -f "$deepep_path" ]] || log "WARN: deepep-config not found: ${deepep_path}"
   prefill_args["--deepep-config"]="${deepep_path}"
   decode_args["--deepep-config"]="${deepep_path}"
 
@@ -211,6 +395,15 @@ _parse_cli_pairs() {
   done
 }
 
+_set_worker_disaggregation_modes() {
+  if [[ "${prefill_config["num-nodes"]:-0}" -gt 0 ]]; then
+    prefill_args["--disaggregation-mode"]="prefill"
+    decode_args["--disaggregation-mode"]="decode"
+  else
+    decode_args["--disaggregation-mode"]="agg"
+  fi
+}
+
 _populate_nodelist() {
   local num_nodes="$1"
   local exclude_nodelist="$2"
@@ -316,6 +509,9 @@ _patch_dynamo_args() {
 _patch_section_args() {
   if _is_sglang_dsr1; then
     _apply_sglang_dsr1_section_args
+  else
+    _apply_multinode_role_args prefill
+    _apply_multinode_role_args decode
   fi
 }
 
@@ -333,6 +529,21 @@ _compute_worker_allocation_sglang_dsr1() {
   decode_config["workers-per-node"]=1
 }
 
+_role_gpu_footprint() {
+  local role="$1"
+  local -n role_config="${role}_config"
+  local -n role_args="${role}_args"
+  local data_parallel_size
+  local world_size=$(( role_args["--tensor-parallel-size"] * role_args["--pipeline-parallel-size"] ))
+  data_parallel_size="$(_role_data_parallel_size "$role")"
+
+  if _role_uses_vllm_multinode_dp "$role"; then
+    echo $(( world_size * data_parallel_size / role_config["nodes-per-worker"] ))
+  else
+    echo $(( world_size / role_config["nodes-per-worker"] ))
+  fi
+}
+
 _compute_worker_allocation_vllm() {
   local num_gpus="$(_gpus_per_node)"
 
@@ -341,13 +552,20 @@ _compute_worker_allocation_vllm() {
     exit 1
   fi
 
-  prefill_config["gpus-per-worker"]=$(( prefill_args["--tensor-parallel-size"] * prefill_args["--pipeline-parallel-size"] ))
-  decode_config["gpus-per-worker"]=$(( decode_args["--tensor-parallel-size"] * decode_args["--pipeline-parallel-size"] ))
+  local prefill_nodes_per_worker=${prefill_config["nodes-per-worker"]}
+  local decode_nodes_per_worker=${decode_config["nodes-per-worker"]}
+  prefill_config["gpus-per-worker"]="$(_role_gpu_footprint prefill)"
+  decode_config["gpus-per-worker"]="$(_role_gpu_footprint decode)"
 
   if [[ ${prefill_config["gpus-per-worker"]} -eq 0 ]] || [[ ${decode_config["gpus-per-worker"]} -eq 0 ]]; then
     log "ERROR: Invalid TP/PP configuration"
     exit 1
   fi
+  if [[ ${prefill_config["gpus-per-worker"]} -gt $num_gpus ]] \
+    || [[ ${decode_config["gpus-per-worker"]} -gt $num_gpus ]]; then
+    log "ERROR: Worker GPU footprint exceeds the ${num_gpus} GPU(s) available on this node"
+    exit 1
+  fi
 
   decode_config["gpu-offset"]=0
   prefill_config["gpu-offset"]=0
@@ -362,16 +580,24 @@ _compute_worker_allocation_vllm() {
     prefill_config["workers-per-node"]=1
     prefill_config["gpu-offset"]=${decode_config["gpus-per-worker"]}
   else
-    if [[ "${prefill_config["multiple-workers-per-node"],,}" != "true" ]]; then
+    if [[ "$prefill_nodes_per_worker" -eq 1 && "${prefill_config["multiple-workers-per-node"],,}" != "true" ]]; then
       prefill_config["gpus-per-worker"]=$num_gpus
     fi
 
-    if [[ "${decode_config["multiple-workers-per-node"],,}" != "true" ]]; then
+    if [[ "$decode_nodes_per_worker" -eq 1 && "${decode_config["multiple-workers-per-node"],,}" != "true" ]]; then
       decode_config["gpus-per-worker"]=$num_gpus
     fi
 
-    prefill_config["workers-per-node"]=$(( num_gpus / prefill_config["gpus-per-worker"] ))
-    decode_config["workers-per-node"]=$(( num_gpus / decode_config["gpus-per-worker"] ))
+    if [[ "$prefill_nodes_per_worker" -gt 1 ]]; then
+      prefill_config["workers-per-node"]=1
+    else
+      prefill_config["workers-per-node"]=$(( num_gpus / prefill_config["gpus-per-worker"] ))
+    fi
+    if [[ "$decode_nodes_per_worker" -gt 1 ]]; then
+      decode_config["workers-per-node"]=1
+    else
+      decode_config["workers-per-node"]=$(( num_gpus / decode_config["gpus-per-worker"] ))
+    fi
   fi
 
   log "DECODE: num GPUs: $num_gpus, GPUs per worker: ${decode_config["gpus-per-worker"]}"
@@ -421,7 +647,10 @@ _dump_args() {
 function parse_args()
 {
   _parse_cli_pairs "$@"
+  _set_worker_disaggregation_modes
+  _normalize_worker_topology
   _set_nodelists
+  _validate_worker_topology
   _patch_dynamo_args
 
   _patch_section_args
@@ -498,14 +727,6 @@ exit_on_error() {
   fi
 }
 
-_total_workers_prefill() {
-  echo $(( prefill_config["num-nodes"] * prefill_config["workers-per-node"] ))
-}
-
-_total_workers_decode() {
-  echo $(( decode_config["num-nodes"] * decode_config["workers-per-node"] ))
-}
-
 _count_initialized_prefill() {
   grep -i -l -E "${prefill_config["worker-initialized-regex"]}" "${RESULTS_DIR}"/dynamo_*prefill* 2>/dev/null | wc -l
 }
@@ -514,21 +735,17 @@ _count_initialized_decode() {
   grep -i -l -E "${decode_config["worker-initialized-regex"]}" "${RESULTS_DIR}"/dynamo_*decode* 2>/dev/null | wc -l
 }
 
-_expected_ready_prefill() {
-  if _is_sglang_dsr1; then
-    echo 1
-  else
-    echo "$(_total_workers_prefill)"
-  fi
-}
+_expected_ready_workers() {
+  local role="$1"
+  local -n role_config="${role}_config"
 
-_expected_ready_decode() {
-  if _is_sglang_dsr1; then
-    echo 1
+  if _role_uses_vllm_multinode_dp "$role"; then
+    echo $(( role_config["num-nodes"] * role_config["workers-per-node"] ))
   else
-    echo "$(_total_workers_decode)"
+    echo $(( (role_config["num-nodes"] / role_config["nodes-per-worker"]) * role_config["workers-per-node"] ))
   fi
 }
+
 _gpu_list_for_worker() {
   local per_worker=$1
   local idx=$2
@@ -596,8 +813,13 @@ _init_runtime_env() {
     export HF_HOME="${HUGGINGFACE_HOME}"
     hf cache scan || echo "HF cache scan failed"
   fi
-  export NATS_SERVER="nats://${dynamo_args["frontend-node"]}:${dynamo_args["nats-port"]}"
-  export ETCD_ENDPOINTS="http://${dynamo_args["frontend-node"]}:${dynamo_args["etcd-port"]}"
+  local frontend_host="${dynamo_args["frontend-node"]}"
+  export NATS_SERVER="${NATS_SERVER:-nats://${frontend_host}:${dynamo_args["nats-port"]}}"
+  export ETCD_ENDPOINTS="${ETCD_ENDPOINTS:-http://${frontend_host}:${dynamo_args["etcd-port"]}}"
+  export DYN_DISCOVERY_BACKEND="${DYN_DISCOVERY_BACKEND:-etcd}"
+  # Dynamo 1.3.1 defaults to ZMQ. Keep CloudAI's existing NATS service
+  # contract explicit while allowing users to opt into another event plane.
+  export DYN_EVENT_PLANE="${DYN_EVENT_PLANE:-nats}"
   export UCX_LOG_FILE="${RESULTS_DIR}/ucx_log_%h.log"
 
   # If KVBM is enabled and leader ports are not explicitly provided, derive a
@@ -772,6 +994,13 @@ validate_environment() {
     _check_free_port_or_die "ingress http" "${dynamo_args["port"]}"
   fi
 
+  if _is_decode_node && _role_is_multinode decode && _role_is_group_leader decode; then
+    _check_free_port_or_die "decode multinode rendezvous" "${dynamo_args["decode-port"]}"
+  fi
+  if _is_prefill_node && _role_is_multinode prefill && _role_is_group_leader prefill; then
+    _check_free_port_or_die "prefill multinode rendezvous" "${dynamo_args["prefill-port"]}"
+  fi
+
   # Decode-node checks for KVBM leader ports (one pub/ack pair per worker).
   if _is_decode_node && _has_connector "kvbm"; then
     local workers_per_node=${decode_config["workers-per-node"]}
@@ -822,10 +1051,11 @@ function mark_failed()
 
 function launch_etcd()
 {
-  log "Launching etcd with cmd: ${dynamo_args["etcd-cmd"]} --listen-client-urls http://0.0.0.0:${dynamo_args["etcd-port"]} --advertise-client-urls http://0.0.0.0:${dynamo_args["etcd-port"]}"
+  local advertise_url="http://${dynamo_args["frontend-node"]}:${dynamo_args["etcd-port"]}"
+  log "Launching etcd with cmd: ${dynamo_args["etcd-cmd"]} --listen-client-urls http://0.0.0.0:${dynamo_args["etcd-port"]} --advertise-client-urls ${advertise_url}"
   ${dynamo_args["etcd-cmd"]} \
     --listen-client-urls http://0.0.0.0:${dynamo_args["etcd-port"]} \
-    --advertise-client-urls http://0.0.0.0:${dynamo_args["etcd-port"]} \
+    --advertise-client-urls "${advertise_url}" \
     > ${RESULTS_DIR}/etcd.log 2>&1
 }
 
@@ -982,9 +1212,70 @@ launch_sgl_http_server() {
   fi
 }
 
+_wait_for_tcp() {
+  local host="$1"
+  local port="$2"
+  local timeout="${3:-1800}"
+  log "Waiting up to ${timeout}s for ${host}:${port}"
+  python3 - "$host" "$port" "$timeout" <<'PY'
+import socket
+import sys
+import time
+
+host, port, timeout = sys.argv[1], int(sys.argv[2]), int(sys.argv[3])
+deadline = time.monotonic() + timeout
+while time.monotonic() < deadline:
+    try:
+        with socket.create_connection((host, port), timeout=2):
+            sys.exit(0)
+    except OSError:
+        time.sleep(2)
+print(f"Timed out waiting for {host}:{port}", file=sys.stderr)
+sys.exit(1)
+PY
+}
+
+_wait_for_role_leader() {
+  local role="$1"
+  if ! _is_vllm || ! _role_is_multinode "$role" || _role_uses_vllm_multinode_dp "$role" \
+    || _role_is_group_leader "$role"; then
+    return
+  fi
+
+  local -n role_args="${role}_args"
+  local leader_host leader_address leader_port
+  leader_host="$(_role_group_leader "$role")"
+  leader_address="${role_args["--master-addr"]:-$leader_host}"
+  leader_port="${role_args["--master-port"]:-${dynamo_args["${role}-port"]}}"
+  _wait_for_tcp "$leader_address" "$leader_port" "${DYNAMO_MULTINODE_START_TIMEOUT:-1800}" || {
+    mark_failed "${role} worker timed out waiting for multinode leader ${leader_host}"
+    return 1
+  }
+}
+
+_monitor_worker_processes() {
+  local role="$1"
+  shift
+  local -a pids=("$@")
+  local pid status
+
+  while [[ ! -f "$DONE_MARKER" && ! -f "$FATAL_ERROR_MARKER" ]]; do
+    for pid in "${pids[@]}"; do
+      if ! kill -0 "$pid" 2>/dev/null; then
+        wait "$pid"
+        status=$?
+        mark_failed "${role} worker process ${pid} exited unexpectedly with status ${status}"
+        return 1
+      fi
+    done
+    sleep 5
+  done
+}
+
 function launch_decode()
 {
   wait_for_etcd
+  _wait_for_role_leader decode || return $?
 
   local workers_per_node=${decode_config["workers-per-node"]}
   local tp_size=${decode_args["--tensor-parallel-size"]}
@@ -995,7 +1286,16 @@ function launch_decode()
   local base_system_port=${DYN_SYSTEM_PORT:-9090}
   local kvbm_port_stride=2
   local side_channel_host
-  side_channel_host="$(_current_node_ip)"
+  local -a worker_pids=()
+  local -a launch_only_args=()
+  local -a extra_args=()
+  read -ra extra_args <<< "${decode_config["extra-args"]:-}"
+  side_channel_host="${VLLM_NIXL_SIDE_CHANNEL_HOST:-$(_current_node_name)}"
+  if _role_uses_vllm_multinode_dp decode; then
+    launch_only_args+=("--data-parallel-hybrid-lb")
+  elif _role_is_vllm_headless decode; then
+    launch_only_args+=("--headless")
+  fi
   log "Launching $workers_per_node decode worker(s) with unique port ranges"
 
   for i in $(seq 0 $(( $workers_per_node - 1 ))); do
@@ -1016,11 +1316,11 @@ function launch_decode()
     # multi-word values (e.g. --cmd "genai-perf profile") through word splitting.
     local -a args_arr=()
     for key in "${!decode_args[@]}"; do
-      args_arr+=($key "${decode_args[$key]}")
+      args_arr+=("$key" "${decode_args[$key]}")
     done
 
     log "Launching decode worker $i on GPUs $gpu_list (NIXL host: $side_channel_host, NIXL port: $nixl_port, KV event port: $kv_event_port, KVBM pub/ack: $kvbm_pub_port/$kvbm_ack_port)"
-    log "Decode cmd: ${decode_config["cmd"]} ${args_arr[*]} ${decode_config["extra-args"]}"
+    log "Decode cmd: ${decode_config["cmd"]} ${args_arr[*]} ${extra_args[*]}"
     CUDA_VISIBLE_DEVICES=$gpu_list \
       DYN_SYSTEM_PORT=$system_port \
       VLLM_NIXL_SIDE_CHANNEL_HOST="$side_channel_host" \
@@ -1029,9 +1329,12 @@ function launch_decode()
       DYN_KVBM_LEADER_ZMQ_PUB_PORT=$kvbm_pub_port \
       DYN_KVBM_LEADER_ZMQ_ACK_PORT=$kvbm_ack_port \
       ${decode_config["cmd"]} \
-      ${args_arr[@]} \
-      ${decode_config["extra-args"]} > $log_file 2>&1 &
+      "${args_arr[@]}" \
+      "${launch_only_args[@]}" \
+      "${extra_args[@]}" > $log_file 2>&1 &
+    worker_pids+=("$!")
   done
+  _monitor_worker_processes decode "${worker_pids[@]}"
 }
 
 function wait_for_etcd()
@@ -1046,6 +1349,7 @@ function wait_for_etcd()
 function launch_prefill()
 {
   wait_for_etcd
+  _wait_for_role_leader prefill || return $?
 
   local workers_per_node=${prefill_config["workers-per-node"]}
   local tp_size=${prefill_args["--tensor-parallel-size"]}
@@ -1057,7 +1361,16 @@ function launch_prefill()
   local kvbm_port_stride=2
   local gpu_offset=${prefill_config["gpu-offset"]:-0}
   local side_channel_host
-  side_channel_host="$(_current_node_ip)"
+  local -a worker_pids=()
+  local -a launch_only_args=()
+  local -a extra_args=()
+  read -ra extra_args <<< "${prefill_config["extra-args"]:-}"
+  side_channel_host="${VLLM_NIXL_SIDE_CHANNEL_HOST:-$(_current_node_name)}"
+  if _role_uses_vllm_multinode_dp prefill; then
+    launch_only_args+=("--data-parallel-hybrid-lb")
+  elif _role_is_vllm_headless prefill; then
+    launch_only_args+=("--headless")
+  fi
 
   if [[ "${SHARED_NODE_DISAGG}" == "true" ]]; then
     local decode_workers=${decode_config["workers-per-node"]}
@@ -1090,11 +1403,11 @@ function launch_prefill()
     # multi-word values (e.g. --cmd "genai-perf profile") through word splitting.
     local -a args_arr=()
     for key in "${!prefill_args[@]}"; do
-      args_arr+=($key "${prefill_args[$key]}")
+      args_arr+=("$key" "${prefill_args[$key]}")
     done
 
     log "Launching prefill worker $i on GPUs $gpu_list (NIXL host: $side_channel_host, NIXL port: $nixl_port, KV event port: $kv_event_port, KVBM pub/ack: $kvbm_pub_port/$kvbm_ack_port)"
-    log "Prefill cmd: ${prefill_config["cmd"]} ${args_arr[*]} ${prefill_config["extra-args"]}"
+    log "Prefill cmd: ${prefill_config["cmd"]} ${args_arr[*]} ${extra_args[*]}"
     CUDA_VISIBLE_DEVICES=$gpu_list \
       DYN_SYSTEM_PORT=$system_port \
       VLLM_NIXL_SIDE_CHANNEL_HOST="$side_channel_host" \
@@ -1103,15 +1416,18 @@ function launch_prefill()
       DYN_KVBM_LEADER_ZMQ_PUB_PORT=$kvbm_pub_port \
       DYN_KVBM_LEADER_ZMQ_ACK_PORT=$kvbm_ack_port \
       ${prefill_config["cmd"]} \
-      ${args_arr[@]} \
-      ${prefill_config["extra-args"]} > $log_file 2>&1 &
+      "${args_arr[@]}" \
+      "${launch_only_args[@]}" \
+      "${extra_args[@]}" > $log_file 2>&1 &
+    worker_pids+=("$!")
   done
+  _monitor_worker_processes prefill "${worker_pids[@]}"
 }
 
 function wait_for_dynamo_frontend()
 {
-  local want_prefill=$(_expected_ready_prefill)
-  local want_decode=$(_expected_ready_decode)
+  local want_prefill=$(_expected_ready_workers prefill)
+  local want_decode=$(_expected_ready_workers decode)
 
   while :; do
     local have_prefill=$(_count_initialized_prefill)
@@ -1153,25 +1469,35 @@ _resolve_aiperf_server_metrics_urls() {
   local base_system_port=${DYN_SYSTEM_PORT:-9090}
   local decode_workers_per_node=${decode_config["workers-per-node"]:-1}
   local prefill_workers_per_node=${prefill_config["workers-per-node"]:-1}
+  local decode_nodes_per_worker=${decode_config["nodes-per-worker"]:-1}
+  local prefill_nodes_per_worker=${prefill_config["nodes-per-worker"]:-1}
   local prefill_system_port_offset=0
   local IFS_SAVE="$IFS"
-  local node i
+  local node i node_index
 
   if [[ "${SHARED_NODE_DISAGG}" == "true" ]]; then
     prefill_system_port_offset=$decode_workers_per_node
   fi
 
   IFS=','
+  node_index=0
   for node in ${decode_config["node-list"]:-}; do
-    for i in $(seq 0 $(( decode_workers_per_node - 1 ))); do
-      urls="${urls},http://${node}:$((base_system_port + i))/metrics"
-    done
+    if _role_uses_vllm_multinode_dp decode || [[ $((node_index % decode_nodes_per_worker)) -eq 0 ]]; then
+      for i in $(seq 0 $(( decode_workers_per_node - 1 ))); do
+        urls="${urls},http://${node}:$((base_system_port + i))/metrics"
+      done
+    fi
+    node_index=$((node_index + 1))
   done
 
+  node_index=0
   for node in ${prefill_config["node-list"]:-}; do
-    for i in $(seq 0 $(( prefill_workers_per_node - 1 ))); do
-      urls="${urls},http://${node}:$((base_system_port + prefill_system_port_offset + i))/metrics"
-    done
+    if _role_uses_vllm_multinode_dp prefill || [[ $((node_index % prefill_nodes_per_worker)) -eq 0 ]]; then
+      for i in $(seq 0 $(( prefill_workers_per_node - 1 ))); do
+        urls="${urls},http://${node}:$((base_system_port + prefill_system_port_offset + i))/metrics"
+      done
+    fi
+    node_index=$((node_index + 1))
   done
 
   if [[ "${dynamo_args["dcgm-exporter-enabled"],,}" == "true" ]]; then
@@ -1259,14 +1585,14 @@ function render_lmcache_config()
   _require_cmd python3
 
   local frontend_node="${dynamo_args["frontend-node"]}"
-  local frontend_ip="$(_resolve_host_ip "$frontend_node")"
+  local frontend_address="$frontend_node"
   local storage_cache_dir="$(lmcache_storage_cache_dir)"
   mkdir -p "$storage_cache_dir"
   chmod 755 "$storage_cache_dir"
 
   local rendered_config="${LMCACHE_CONFIG_FILE}.tmp.${SLURM_NODEID:-0}"
   if ! FRONTEND_NODE="$frontend_node" \
-    FRONTEND_IP="$frontend_ip" \
+    FRONTEND_IP="$frontend_address" \
     RESULTS_DIR="$RESULTS_DIR" \
     STORAGE_CACHE_DIR="$storage_cache_dir" \
     python3 - "$LMCACHE_CONFIG_FILE" "$rendered_config" <<'PY'
@@ -1447,12 +1773,7 @@ function main()
     wait_for_etcd
   fi
 
-  # Workers launch BEFORE the ingress: launch_ingress blocks in
-  # wait_for_router, and the router only becomes ready once a worker
-  # registers — on a combined frontend+worker node the old order serialized
-  # the whole ROUTER_START_TIMEOUT (120 s of failing readiness curls) in
-  # front of every worker start. Workers only need etcd/nats (waited above)
-  # and the lmcache config from setup_lmcache; they never talk to the router.
+  # Router readiness depends on worker registration, so launch workers before ingress.
   if _is_decode_node; then
     log "Node ID: $SLURM_NODEID, Role: decode"
     log_node_role "$(_current_node_name)" "decode"
@@ -1478,8 +1799,9 @@ function main()
   wait_for_frontend_marker
 }
 
-log "Starting main"
-main "$@"
-log "Done with main"
-
-perform_exit 0
+if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
+  log "Starting main"
+  main "$@"
+  log "Done with main"
+  perform_exit 0
+fi
diff --git a/src/cloudai/workloads/ai_dynamo/slurm_command_gen_strategy.py b/src/cloudai/workloads/ai_dynamo/slurm_command_gen_strategy.py
index 0c09c1f7b..64b5b71ab 100644
--- a/src/cloudai/workloads/ai_dynamo/slurm_command_gen_strategy.py
+++ b/src/cloudai/workloads/ai_dynamo/slurm_command_gen_strategy.py
@@ -35,6 +35,7 @@
     AIDynamoTestDefinition,
     AIPerf,
     AIPerfPhase,
+    WorkerConfig,
 )
 
 AIPERF_SCRIPT_FILE_NAME = "aiperf.sh"
@@ -59,6 +60,7 @@ def _container_mounts(self) -> list[str]:
     @property
     def final_env_vars(self) -> dict[str, str | list[str]]:
         env_vars = super().final_env_vars
+        env_vars["DYNAMO_NODELIST"] = "$(scontrol show hostname $SLURM_JOB_NODELIST | tr -s '\\n' ',' | sed 's/,$//')"
         if self.td.cmd_args.hicache is not None:
             env_vars["HICACHE_CONFIG_FILE"] = f"{self.CONTAINER_MOUNT_OUTPUT}/{HICACHE_CONFIG_FILE_NAME}"
         if self.td.cmd_args.lmcache is not None:
@@ -175,6 +177,55 @@ def _prepare_hicache_config(self) -> None:
         config = toml.dumps(self.td.cmd_args.hicache)
         (self.test_run.output_path / HICACHE_CONFIG_FILE_NAME).write_text(config)
 
+    def _validate_multinode_worker(self, role: str, worker: WorkerConfig) -> None:
+        if worker.nodes_per_worker is None:
+            return
+        if not isinstance(worker.num_nodes, int) or not isinstance(worker.nodes_per_worker, int):
+            raise ValueError(f"{role} worker topology must be scalar after DSE unrolling")
+
+        nodes_per_worker = worker.nodes_per_worker
+        if nodes_per_worker <= 1:
+            return
+
+        tp = worker.args.tensor_parallel_size
+        pp = worker.args.pipeline_parallel_size
+        dp = worker.args.data_parallel_size
+        if not isinstance(tp, int) or not isinstance(pp, int) or isinstance(dp, list):
+            raise ValueError(f"{role} worker parallelism must be scalar after DSE unrolling")
+
+        world_size = tp * pp
+        data_parallel_size = int(dp or 1)
+        backend = self.td.cmd_args.dynamo.backend
+        is_vllm_multinode_dp = backend == "vllm" and data_parallel_size > 1
+        if backend == "sglang" and data_parallel_size > 1 and not worker.has_extra_arg("--enable-dp-attention"):
+            raise ValueError(f"Multinode SGLang data parallelism for the {role} worker requires --enable-dp-attention")
+        if is_vllm_multinode_dp and data_parallel_size % nodes_per_worker != 0:
+            raise ValueError(
+                f"{role} worker data_parallel_size ({data_parallel_size}) must be divisible by "
+                f"nodes_per_worker ({nodes_per_worker})"
+            )
+        if not is_vllm_multinode_dp and world_size % nodes_per_worker != 0:
+            raise ValueError(
+                f"{role} worker TP*PP ({world_size}) must be divisible by nodes_per_worker ({nodes_per_worker})"
+            )
+        local_world_size = (
+            world_size * (data_parallel_size // nodes_per_worker)
+            if is_vllm_multinode_dp
+            else world_size // nodes_per_worker
+        )
+        gpus_per_node = int(getattr(self.system, "gpus_per_node", 0) or 0)
+        if gpus_per_node and local_world_size > gpus_per_node:
+            raise ValueError(
+                f"{role} worker needs {local_world_size} GPU(s) per node, but the system has {gpus_per_node}"
+            )
+        if self.td.cmd_args.dynamo.backend == "vllm" and worker.args.distributed_executor_backend not in {None, "mp"}:
+            raise ValueError("Multinode Dynamo vLLM currently supports only the mp distributed executor backend")
+
+    def _validate_multinode_workers(self) -> None:
+        dynamo = self.td.cmd_args.dynamo
+        self._validate_multinode_worker("prefill", dynamo.prefill_worker)
+        self._validate_multinode_worker("decode", dynamo.decode_worker)
+
     def _render_aiperf_args(self, args: dict[str, Any]) -> str:
         parts: list[str] = []
         for key, value in args.items():
@@ -407,6 +458,7 @@ def _prepare_aiperf_script(self) -> str | None:
         return f"{self.CONTAINER_MOUNT_OUTPUT}/{AIPERF_SCRIPT_FILE_NAME}"
 
     def _gen_script_args(self, td: AIDynamoTestDefinition) -> List[str]:
+        self._validate_multinode_workers()
         self._prepare_hicache_config()
         self._prepare_lmcache_config()
         aiperf_script = self._prepare_aiperf_script()
diff --git a/tests/ref_data/ai-dynamo.sbatch b/tests/ref_data/ai-dynamo.sbatch
index 353053c57..60c146673 100644
--- a/tests/ref_data/ai-dynamo.sbatch
+++ b/tests/ref_data/ai-dynamo.sbatch
@@ -9,7 +9,7 @@
 #SBATCH --gres=gpu:8
 
 export SLURM_JOB_MASTER_NODE=$(scontrol show hostname $SLURM_JOB_NODELIST | head -n 1)
-
+export DYNAMO_NODELIST=$(scontrol show hostname $SLURM_JOB_NODELIST | tr -s '\n' ',' | sed 's/,$//')
 srun --export=ALL --mpi=pmix -N2 --container-image=nvcr.io/nvidia/ai-dynamo:24.09 --container-mounts=__OUTPUT_DIR__/output:/cloudai_run_results,__INSTALL_DIR__:/cloudai_install,__OUTPUT_DIR__/output,__INSTALL_DIR__/huggingface:/cloudai_install/huggingface,/tmp:/tmp --output=__OUTPUT_DIR__/output/mapping-stdout.txt --error=__OUTPUT_DIR__/output/mapping-stderr.txt bash -c "echo \$(date): \$(hostname):node \${SLURM_NODEID}:rank \${SLURM_PROCID}."
 
 srun --export=ALL --mpi=none -N2 --ntasks=2 --ntasks-per-node=1 --output=__OUTPUT_DIR__/output/metadata/node-%N.toml --error=__OUTPUT_DIR__/output/metadata/nodes.err --container-image=nvcr.io/nvidia/ai-dynamo:24.09 --container-mounts=__OUTPUT_DIR__/output:/cloudai_run_results,__INSTALL_DIR__:/cloudai_install,__OUTPUT_DIR__/output,__INSTALL_DIR__/huggingface:/cloudai_install/huggingface,/tmp:/tmp bash /cloudai_install/slurm-metadata.sh runtime
@@ -78,7 +78,7 @@ srun \
   --user $USER \
   --install-dir /cloudai_install \
   --results-dir /cloudai_run_results \
-  --dynamo-repo /cloudai_install/dynamo__f7e468c7e8ff0d1426db987564e60572167e8464 \
+  --dynamo-repo /cloudai_install/dynamo__a49702e4432e7fa43cbc88175bddb31604340f19 \
   --hf-home /cloudai_install/huggingface \
   --workloads aiperf.sh \
   --failure-marker /cloudai_run_results/failure-marker.txt \
@@ -97,7 +97,7 @@ srun \
   --dynamo-nats-port "4222" \
   --dynamo-dcgm-exporter-enabled "True" \
   --dynamo-dcgm-exporter-port "9501" \
-  --prefill-cmd "python3 -m dynamo.vllm --is-prefill-worker" \
+  --prefill-cmd "python3 -m dynamo.vllm" \
   --prefill-worker-initialized-regex "VllmWorker.*has.been.initialized" \
   --prefill-multiple-workers-per-node "False" \
   --prefill-num-nodes "1" \
diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py
index 080799d77..0a68a8750 100644
--- a/tests/test_acceptance.py
+++ b/tests/test_acceptance.py
@@ -535,7 +535,7 @@ def test_req(request, slurm_system: SlurmSystem, partial_tr: partial[TestRun]) -
                             workspace_path="/workspace",
                             dcgm_exporter=DCGMExporter(enabled=True, port=9501),
                             prefill_worker=WorkerConfig(
-                                cmd="python3 -m dynamo.vllm --is-prefill-worker",
+                                cmd="python3 -m dynamo.vllm",
                                 worker_initialized_regex="VllmWorker.*has.been.initialized",
                                 **{
                                     "num-nodes": 1,
diff --git a/tests/workloads/ai_dynamo/test_command_gen_strategy_slurm.py b/tests/workloads/ai_dynamo/test_command_gen_strategy_slurm.py
index 8ef95381e..076379cec 100644
--- a/tests/workloads/ai_dynamo/test_command_gen_strategy_slurm.py
+++ b/tests/workloads/ai_dynamo/test_command_gen_strategy_slurm.py
@@ -74,7 +74,7 @@ def cmd_args() -> AIDynamoCmdArgs:
             model="model",
             workspace_path="/workspace",
             prefill_worker=WorkerConfig(
-                cmd="python3 -m dynamo.vllm --is-prefill-worker",
+                cmd="python3 -m dynamo.vllm",
                 worker_initialized_regex="VllmWorker.*has.been.initialized",
                 **{
                     "num-nodes": 1,
@@ -164,6 +164,14 @@ def test_container_mounts(strategy: AIDynamoSlurmCommandGenStrategy, test_run: T
     assert mounts == expected
 
 
+def test_final_env_vars_populates_dynamo_nodelist(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
+    strategy.test_run.test.extra_env_vars["DYNAMO_NODELIST"] = "stale"
+
+    assert strategy.final_env_vars["DYNAMO_NODELIST"] == (
+        "$(scontrol show hostname $SLURM_JOB_NODELIST | tr -s '\\n' ',' | sed 's/,$//')"
+    )
+
+
 def test_installables_include_top_level_git_repos(cmd_args: AIDynamoCmdArgs) -> None:
     repo = GitRepo(url="https://github.com/example/custom-tools.git", commit="main")
     tdef = AIDynamoTestDefinition(
@@ -603,6 +611,138 @@ def test_constraint_allows_separate_node_roles_using_all_node_gpus(
     assert td.constraint_check(test_run, slurm_system)
 
 
+def test_constraint_allows_multinode_worker_using_group_capacity(slurm_system: SlurmSystem, test_run: TestRun) -> None:
+    slurm_system.gpus_per_node = 4
+    td = cast(AIDynamoTestDefinition, test_run.test)
+    for worker in (td.cmd_args.dynamo.prefill_worker, td.cmd_args.dynamo.decode_worker):
+        worker.num_nodes = 2
+        worker.nodes_per_worker = 2
+        worker.args.tensor_parallel_size = 8
+        worker.args.pipeline_parallel_size = 1
+    test_run.num_nodes = 4
+    test_run.nodes = ["n0", "n1", "n2", "n3"]
+    test_run.num_nodes_explicit = True
+
+    assert td.constraint_check(test_run, slurm_system)
+
+
+def test_constraint_allows_vllm_multinode_dp_using_local_rank_capacity(
+    slurm_system: SlurmSystem, test_run: TestRun
+) -> None:
+    slurm_system.gpus_per_node = 4
+    td = cast(AIDynamoTestDefinition, test_run.test)
+    for worker in (td.cmd_args.dynamo.prefill_worker, td.cmd_args.dynamo.decode_worker):
+        worker.num_nodes = 2
+        worker.nodes_per_worker = 2
+        worker.args.tensor_parallel_size = 1
+        worker.args.pipeline_parallel_size = 1
+        worker.args.data_parallel_size = 8
+    test_run.num_nodes = 4
+    test_run.nodes = ["n0", "n1", "n2", "n3"]
+    test_run.num_nodes_explicit = True
+
+    assert td.constraint_check(test_run, slurm_system)
+
+
+def test_constraint_rejects_multinode_vllm_ray(slurm_system: SlurmSystem, test_run: TestRun) -> None:
+    td = cast(AIDynamoTestDefinition, test_run.test)
+    worker = td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.args.tensor_parallel_size = 8
+    worker.args.distributed_executor_backend = "ray"
+
+    assert not td.constraint_check(test_run, slurm_system)
+
+
+def test_constraint_rejects_multinode_sglang_dp_without_dp_attention(
+    slurm_system: SlurmSystem, test_run: TestRun
+) -> None:
+    td = cast(AIDynamoTestDefinition, test_run.test)
+    td.cmd_args.dynamo.backend = "sglang"
+    worker = td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.args.tensor_parallel_size = 8
+    worker.args.data_parallel_size = 8
+
+    assert not td.constraint_check(test_run, slurm_system)
+
+
+def test_multinode_worker_rejects_unbalanced_world_size(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
+    worker = strategy.td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.args.tensor_parallel_size = 15
+
+    with pytest.raises(ValueError, match=r"TP\*PP \(15\) must be divisible"):
+        strategy._gen_script_args(strategy.td)
+
+
+def test_multinode_vllm_dp_rejects_unbalanced_rank_placement(
+    strategy: AIDynamoSlurmCommandGenStrategy,
+) -> None:
+    worker = strategy.td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.args.tensor_parallel_size = 1
+    worker.args.data_parallel_size = 7
+
+    with pytest.raises(ValueError, match=r"data_parallel_size \(7\) must be divisible"):
+        strategy._gen_script_args(strategy.td)
+
+
+def test_multinode_sglang_dp_accepts_combined_list_extra_args(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
+    strategy.td.cmd_args.dynamo.backend = "sglang"
+    worker = strategy.td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.extra_args = ["--trust-remote-code --enable-dp-attention"]
+    worker.args.tensor_parallel_size = 8
+    worker.args.data_parallel_size = 8
+
+    strategy._gen_script_args(strategy.td)
+
+
+def test_multinode_sglang_dp_requires_dp_attention(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
+    strategy.td.cmd_args.dynamo.backend = "sglang"
+    worker = strategy.td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.args.tensor_parallel_size = 8
+    worker.args.data_parallel_size = 8
+
+    with pytest.raises(ValueError, match="requires --enable-dp-attention"):
+        strategy._gen_script_args(strategy.td)
+
+
+def test_multinode_vllm_worker_rejects_ray(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
+    worker = strategy.td.cmd_args.dynamo.decode_worker
+    worker.num_nodes = 2
+    worker.nodes_per_worker = 2
+    worker.args.tensor_parallel_size = 16
+    worker.args.distributed_executor_backend = "ray"
+
+    with pytest.raises(ValueError, match="only the mp distributed executor"):
+        strategy._gen_script_args(strategy.td)
+
+
+def test_multinode_worker_topology_is_forwarded_to_runtime(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
+    prefill = strategy.td.cmd_args.dynamo.prefill_worker
+    decode = strategy.td.cmd_args.dynamo.decode_worker
+    prefill.num_nodes = 2
+    prefill.nodes_per_worker = 2
+    prefill.args.tensor_parallel_size = 16
+    decode.num_nodes = 2
+    decode.nodes_per_worker = 2
+    decode.args.tensor_parallel_size = 16
+
+    args = strategy._gen_script_args(strategy.td)
+
+    assert '--prefill-nodes-per-worker "2"' in args
+    assert '--decode-nodes-per-worker "2"' in args
+
+
 def test_aiperf_phase_roundtrip_does_not_emit_default_report_name(strategy: AIDynamoSlurmCommandGenStrategy) -> None:
     td = cast(AIDynamoTestDefinition, strategy.test_run.test)
     td.cmd_args.workloads = "aiperf.sh"
diff --git a/tests/workloads/ai_dynamo/test_report_gen_strategy.py b/tests/workloads/ai_dynamo/test_report_gen_strategy.py
index f81e9a2dd..2421a0c96 100644
--- a/tests/workloads/ai_dynamo/test_report_gen_strategy.py
+++ b/tests/workloads/ai_dynamo/test_report_gen_strategy.py
@@ -31,7 +31,7 @@
     WorkerBaseArgs,
     WorkerConfig,
 )
-from cloudai.workloads.ai_dynamo.ai_dynamo import parse_aiperf_accuracy
+from cloudai.workloads.ai_dynamo.ai_dynamo import parse_aiperf_accuracy, parse_aiperf_request_count
 from cloudai.workloads.ai_dynamo.report_generation_strategy import AIDynamoReportGenerationStrategy
 
 
@@ -81,7 +81,7 @@ def ai_dynamo_tr(tmp_path: Path) -> TestRun:
             workloads="genai_perf.sh",
             dynamo=AIDynamoArgs(
                 prefill_worker=WorkerConfig(
-                    cmd="python3 -m dynamo.vllm --is-prefill-worker",
+                    cmd="python3 -m dynamo.vllm",
                     worker_initialized_regex="VllmWorker.*has.been.initialized",
                     args=WorkerBaseArgs(),
                 ),
@@ -112,7 +112,7 @@ def ai_dynamo_aiperf_tr(tmp_path: Path) -> TestRun:
             workloads="aiperf.sh",
             dynamo=AIDynamoArgs(
                 prefill_worker=WorkerConfig(
-                    cmd="python3 -m dynamo.vllm --is-prefill-worker",
+                    cmd="python3 -m dynamo.vllm",
                     worker_initialized_regex="VllmWorker.*has.been.initialized",
                     args=WorkerBaseArgs(),
                 ),
@@ -137,7 +137,7 @@ def ai_dynamo_aiperf_with_split_accuracy_tr(tmp_path: Path) -> TestRun:
             workloads="aiperf.sh",
             dynamo=AIDynamoArgs(
                 prefill_worker=WorkerConfig(
-                    cmd="python3 -m dynamo.vllm --is-prefill-worker",
+                    cmd="python3 -m dynamo.vllm",
                     worker_initialized_regex="VllmWorker.*has.been.initialized",
                     args=WorkerBaseArgs(),
                 ),
@@ -164,7 +164,7 @@ def ai_dynamo_genai_perf_with_split_accuracy_tr(tmp_path: Path) -> TestRun:
             workloads="genai_perf.sh",
             dynamo=AIDynamoArgs(
                 prefill_worker=WorkerConfig(
-                    cmd="python3 -m dynamo.vllm --is-prefill-worker",
+                    cmd="python3 -m dynamo.vllm",
                     worker_initialized_regex="VllmWorker.*has.been.initialized",
                     args=WorkerBaseArgs(),
                 ),
@@ -264,6 +264,17 @@ def test_was_run_successful_with_split_aiperf_accuracy(
     assert result.is_successful is True
 
 
+def test_was_run_successful_rejects_aiperf_report_with_only_errors(ai_dynamo_aiperf_tr: TestRun) -> None:
+    test_def = ai_dynamo_aiperf_tr.test
+    (ai_dynamo_aiperf_tr.output_path / "aiperf_report.csv").write_text(
+        "Metric,Value\nError Request Count,4.00\nTotal Error Input Sequence Length (tokens),1200.00\n"
+    )
+
+    result = test_def.was_run_successful(ai_dynamo_aiperf_tr)
+
+    assert result.is_successful is False
+
+
 def test_was_run_successful_with_genai_perf_and_split_aiperf_accuracy(
     ai_dynamo_genai_perf_with_split_accuracy_tr: TestRun,
 ) -> None:
@@ -297,6 +308,13 @@ def test_parse_aiperf_accuracy_from_artifact_dir(tmp_path: Path) -> None:
     assert parse_aiperf_accuracy(tmp_path) == 0.35
 
 
+def test_parse_aiperf_request_count(tmp_path: Path) -> None:
+    report = tmp_path / "aiperf_report.csv"
+    report.write_text(get_aiperf_csv_content(), encoding="utf-8")
+
+    assert parse_aiperf_request_count(report) == 50
+
+
 def test_parse_aiperf_accuracy_from_split_accuracy_artifact_dir(tmp_path: Path) -> None:
     artifact_dir = tmp_path / "aiperf_accuracy_artifacts"
     artifact_dir.mkdir()