diff --git a/.buildkite/test_areas/distributed.yaml b/.buildkite/test_areas/distributed.yaml
index 06a0b5212eeb..3490892d8013 100644
--- a/.buildkite/test_areas/distributed.yaml
+++ b/.buildkite/test_areas/distributed.yaml
@@ -121,11 +121,14 @@ steps:
   - vllm/v1/engine/llm_engine.py
   - vllm/v1/executor/uniproc_executor.py
   - vllm/v1/worker/gpu_worker.py
+  - tests/distributed/test_mnnvl_alltoall.py
+
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
   - export NCCL_CUMEM_HOST_ENABLE=0
   # test with torchrun tp=2 and dp=4 with ep
   - torchrun --nproc-per-node=8 ../examples/offline_inference/torchrun_dp_example.py --tp-size=2 --pp-size=1 --dp-size=4 --enable-ep
+  - pytest -v -s tests/distributed/test_mnnvl_alltoall.py
 
 - label: Distributed Tests (4 GPUs)(A100)
   device: a100
diff --git a/tests/distributed/test_mnnvl_alltoall.py b/tests/distributed/test_mnnvl_alltoall.py
new file mode 100644
index 000000000000..d53d95a327d4
--- /dev/null
+++ b/tests/distributed/test_mnnvl_alltoall.py
@@ -0,0 +1,1095 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Direct unit and component tests for MNNVL AllToAll operations.
+requires container ran w/ docker run ... --cap-add=SYS_PTRACE ...
+Run `pytest tests/distributed/test_mnnvl_alltoall.py`.
+"""
+
+import os
+import subprocess
+
+import pytest
+import torch
+
+from vllm.distributed import get_ep_group
+from vllm.utils.flashinfer import has_flashinfer_all2all
+
+from ..utils import init_test_distributed_environment
+
+# Skip all tests if FlashInfer alltoall is not available
+pytestmark = pytest.mark.skipif(
+    not has_flashinfer_all2all(),
+    reason="FlashInfer alltoall not available",
+)
+
+
+# Simple placeholder class for attention metadata in tests
+class PlaceholderAttnMetadata:
+    """Placeholder attention metadata for testing."""
+
+    def __init__(self):
+        self.dp_metadata = None
+
+
+def has_sys_ptrace_capability() -> bool:
+    """
+    Check if the process has SYS_PTRACE capability.
+
+    MNNVL (Multi-Node NVLink) requires SYS_PTRACE to share memory file descriptors
+    between processes using pidfd_getfd() system call.
+
+    Returns:
+        True if SYS_PTRACE is available, False otherwise.
+    """
+    try:
+        # Try to check capabilities using capsh if available
+        result = subprocess.run(
+            ["capsh", "--print"], capture_output=True, text=True, timeout=5
+        )
+        if result.returncode == 0 and "cap_sys_ptrace" in result.stdout.lower():
+            return True
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        pass
+
+    # Alternative check: try to read /proc/self/status
+    try:
+        with open("/proc/self/status") as f:
+            for line in f:
+                if line.startswith("CapEff:"):
+                    # SYS_PTRACE is capability bit 19 (0x80000 = 1 << 19)
+                    cap_eff = int(line.split()[1], 16)
+                    # Check if bit 19 is set
+                    return bool(cap_eff & (1 << 19))
+    except Exception:
+        pass
+
+    # If we can't determine, assume it's not available in container environments
+    # Check if we're in a container; if not, assume it's available
+    return not (os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"))
+
+
+def test_flashinfer_all2all_import():
+    """Test that we can import FlashInfer alltoall components."""
+    try:
+        from flashinfer.comm import Mapping
+        from flashinfer.comm.mnnvl import MnnvlConfig
+        from flashinfer.comm.trtllm_alltoall import MnnvlMoe
+
+        from vllm.distributed.device_communicators.all2all import (
+            FlashInferAllToAllManager,
+        )
+        from vllm.distributed.device_communicators.mnnvl_compat import (
+            CustomCommunicator,
+        )
+
+        assert Mapping is not None
+        assert MnnvlConfig is not None
+        assert MnnvlMoe is not None
+        assert FlashInferAllToAllManager is not None
+        assert CustomCommunicator is not None
+    except ImportError as e:
+        pytest.fail(f"Failed to import FlashInfer alltoall components: {e}")
+
+
+def run_multi_gpu_test(rank: int, world_size: int, port: str, test_func):
+    """Helper to run a test function in a multi-GPU distributed environment."""
+    # Remove CUDA_VISIBLE_DEVICES to allow access to all GPUs
+    os.environ.pop("CUDA_VISIBLE_DEVICES", None)
+
+    # Set device for this rank
+    device = torch.device(f"cuda:{rank}")
+    torch.cuda.set_device(device)
+
+    # Initialize distributed environment
+    # Use world_size for tp to create multi-process setup
+    init_test_distributed_environment(world_size, 1, rank, port)
+
+    # Verify multi-GPU setup
+    assert torch.distributed.is_initialized()
+    assert torch.distributed.get_world_size() == world_size
+    assert torch.distributed.get_rank() == rank
+
+    print(
+        f"\n[Rank {rank}] GPU: {torch.cuda.current_device()}, "
+        f"World size: {torch.distributed.get_world_size()}"
+    )
+
+    # Run the actual test
+    test_func(rank, world_size)
+
+    # Synchronize before cleanup
+    torch.distributed.barrier()
+
+
+def manager_initialization_worker(rank: int, world_size: int):
+    """Worker function for testing FlashInferAllToAllManager initialization."""
+    from vllm.distributed.device_communicators.all2all import (
+        FlashInferAllToAllManager,
+    )
+
+    # Get CPU group from EP
+    cpu_group = get_ep_group().cpu_group
+
+    # Create manager
+    manager = FlashInferAllToAllManager(cpu_group)
+
+    # Verify multi-GPU properties
+    print(
+        f"[Rank {rank}] Manager rank: {manager.rank}, world_size: {manager.world_size}"
+    )
+    assert manager is not None
+    assert manager.rank == rank
+    assert manager.world_size == world_size
+    assert not manager.initialized
+
+    # Test workspace initialization - should work with world_size > 1
+    manager.initialize(
+        world_size=world_size,
+        rank=rank,
+        gpus_per_node=torch.cuda.device_count(),
+    )
+
+    assert manager.initialized
+    assert manager.workspace_tensor is not None
+    assert manager.prepare_workspace_tensor is not None
+    assert manager.mapping is not None
+
+    print(f"[Rank {rank}] Manager initialized successfully")
+
+    # Synchronize before cleanup
+    torch.distributed.barrier()
+
+    # Test cleanup
+    manager.cleanup()
+    assert not manager.initialized
+    assert manager.workspace_tensor is None
+    assert manager.prepare_workspace_tensor is None
+
+    print(f"[Rank {rank}] Manager cleanup successful")
+
+
+def workspace_reinitialization_worker(rank: int, world_size: int):
+    """Worker function for testing workspace reinitialization."""
+    from vllm.distributed.device_communicators.all2all import (
+        FlashInferAllToAllManager,
+    )
+
+    cpu_group = get_ep_group().cpu_group
+    manager = FlashInferAllToAllManager(cpu_group)
+
+    # Initialize
+    manager.initialize(
+        world_size=world_size,
+        rank=rank,
+        gpus_per_node=torch.cuda.device_count(),
+    )
+    assert manager.initialized
+    print(f"[Rank {rank}] First initialization complete")
+
+    torch.distributed.barrier()
+
+    # Cleanup
+    manager.cleanup()
+    assert not manager.initialized
+    print(f"[Rank {rank}] Cleanup complete")
+
+    torch.distributed.barrier()
+
+    # Re-initialize
+    manager.initialize(
+        world_size=world_size,
+        rank=rank,
+        gpus_per_node=torch.cuda.device_count(),
+    )
+    assert manager.initialized
+    print(f"[Rank {rank}] Re-initialization complete")
+
+    torch.distributed.barrier()
+
+    manager.cleanup()
+
+
+def ensure_initialized_worker(rank: int, world_size: int):
+    """Worker function for testing ensure_alltoall_workspace_initialized."""
+    from vllm.distributed.device_communicators.all2all import (
+        FlashInferAllToAllManager,
+    )
+
+    cpu_group = get_ep_group().cpu_group
+    manager = FlashInferAllToAllManager(cpu_group)
+
+    # Should not be initialized yet
+    assert not manager.initialized
+
+    # Call ensure - should initialize with world_size > 1
+    result = manager.ensure_alltoall_workspace_initialized()
+    assert result
+    assert manager.initialized
+    print(f"[Rank {rank}] ensure_initialized returned True, manager initialized")
+
+    torch.distributed.barrier()
+
+    # Call again - should return True without re-initializing
+    result = manager.ensure_alltoall_workspace_initialized()
+    assert result
+    assert manager.initialized
+    print(f"[Rank {rank}] ensure_initialized (2nd call) returned True")
+
+    torch.distributed.barrier()
+
+    manager.cleanup()
+
+
+@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Need at least 2 GPUs")
+@pytest.mark.skipif(
+    not has_sys_ptrace_capability(),
+    reason=(
+        "SYS_PTRACE capability required for MNNVL. "
+        "Run container with: docker run --cap-add=SYS_PTRACE"
+    ),
+)
+@pytest.mark.parametrize("world_size", [2])
+def test_flashinfer_alltoall_manager_initialization(world_size: int):
+    """
+    Test FlashInferAllToAllManager initialization with multiple GPUs.
+
+    This test spawns multiple processes (one per GPU) to test actual multi-GPU
+    AllToAll operations. Requires SYS_PTRACE capability for MNNVL memory sharing.
+    """
+    import torch.multiprocessing as mp
+
+    # Use spawn method for CUDA compatibility
+    mp.set_start_method("spawn", force=True)
+
+    port = "12355"
+
+    # Launch multiple processes
+    processes = []
+    for rank in range(world_size):
+        p = mp.Process(
+            target=run_multi_gpu_test,
+            args=(rank, world_size, port, manager_initialization_worker),
+        )
+        p.start()
+        processes.append(p)
+
+    # Wait for all processes
+    for p in processes:
+        p.join()
+        assert p.exitcode == 0, f"Process failed with exit code {p.exitcode}"
+
+
+@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Need at least 2 GPUs")
+@pytest.mark.skipif(
+    not has_sys_ptrace_capability(),
+    reason=(
+        "SYS_PTRACE capability required for MNNVL. "
+        "Run container with: docker run --cap-add=SYS_PTRACE"
+    ),
+)
+@pytest.mark.parametrize("world_size", [2])
+def test_flashinfer_alltoall_workspace_reinitialization(world_size: int):
+    """
+    Test that workspace can be reinitialized with multiple GPUs.
+
+    This test spawns multiple processes to test workspace reinitialization.
+    Requires SYS_PTRACE capability for MNNVL memory sharing.
+    """
+    import torch.multiprocessing as mp
+
+    mp.set_start_method("spawn", force=True)
+
+    port = "12356"
+
+    # Launch multiple processes
+    processes = []
+    for rank in range(world_size):
+        p = mp.Process(
+            target=run_multi_gpu_test,
+            args=(rank, world_size, port, workspace_reinitialization_worker),
+        )
+        p.start()
+        processes.append(p)
+
+    # Wait for all processes
+    for p in processes:
+        p.join()
+        assert p.exitcode == 0, f"Process failed with exit code {p.exitcode}"
+
+
+@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Need at least 2 GPUs")
+@pytest.mark.skipif(
+    not has_sys_ptrace_capability(),
+    reason=(
+        "SYS_PTRACE capability required for MNNVL. "
+        "Run container with: docker run --cap-add=SYS_PTRACE"
+    ),
+)
+@pytest.mark.parametrize("world_size", [2])
+def test_flashinfer_alltoall_ensure_initialized(world_size: int):
+    """
+    Test ensure_alltoall_workspace_initialized with multiple GPUs.
+
+    This test spawns multiple processes to test the ensure_initialized method.
+    Requires SYS_PTRACE capability for MNNVL memory sharing.
+    """
+    import torch.multiprocessing as mp
+
+    mp.set_start_method("spawn", force=True)
+
+    port = "12357"
+
+    # Launch multiple processes
+    processes = []
+    for rank in range(world_size):
+        p = mp.Process(
+            target=run_multi_gpu_test,
+            args=(rank, world_size, port, ensure_initialized_worker),
+        )
+        p.start()
+        processes.append(p)
+
+    # Wait for all processes
+    for p in processes:
+        p.join()
+        assert p.exitcode == 0, f"Process failed with exit code {p.exitcode}"
+
+
+# =============================================================================
+# Data Communication Validation Tests
+# =============================================================================
+# These tests validate that the a2a (all-to-all) backends correctly communicate
+# data between ranks by comparing results against reference implementations.
+# Three levels of validation:
+# 1. Basic data communication - compares AgRs vs Naive backends
+# 2. FlashInfer validation - tests MNNVL a2a backend against reference
+# 3. Deterministic validation - verifies exact data values with known patterns
+# =============================================================================
+
+
+def data_communication_worker(rank: int, world_size: int):
+    """
+    Worker function for testing actual data communication via AllToAll.
+
+    This test validates that the FlashInferAllToAllManager correctly
+    communicates data by comparing against reference backends.
+    """
+    from vllm.config.vllm import VllmConfig
+    from vllm.distributed.device_communicators.all2all import (
+        AgRsAll2AllManager,
+    )
+    from vllm.forward_context import set_forward_context
+
+    # Get CPU group
+    cpu_group = get_ep_group().cpu_group
+
+    # Test dimensions
+    hidden_size = 128
+    num_tokens_per_rank = 32
+    num_experts_per_token = 2
+    num_global_experts = world_size * 4  # 4 experts per rank
+
+    # Create test input data (unique per rank)
+    torch.manual_seed(rank + 42)
+    device = torch.device(f"cuda:{rank}")
+
+    hidden_states = torch.randn(
+        num_tokens_per_rank, hidden_size, device=device, dtype=torch.float16
+    )
+    topk_weights = torch.randn(
+        num_tokens_per_rank, num_experts_per_token, device=device, dtype=torch.float16
+    )
+    topk_ids = torch.randint(
+        0,
+        num_global_experts,
+        (num_tokens_per_rank, num_experts_per_token),
+        device=device,
+        dtype=torch.long,
+    )
+    router_logits = torch.randn(
+        num_tokens_per_rank, num_global_experts, device=device, dtype=torch.float16
+    )
+
+    # Create mock forward context with dp_metadata
+    class MockDPMetadata:
+        def __init__(self, world_size, num_tokens_per_rank):
+            self.world_size = world_size
+            self.num_tokens_per_rank = num_tokens_per_rank
+
+        def cu_tokens_across_sp(self, sp_size):
+            """Cumulative token counts across sequence parallel ranks."""
+            cu_tokens = torch.tensor(
+                [i * self.num_tokens_per_rank for i in range(1, self.world_size + 1)],
+                dtype=torch.int32,
+            )
+            return cu_tokens
+
+        def get_chunk_sizes_across_dp_rank(self):
+            """Get chunk sizes for all ranks."""
+            return [self.num_tokens_per_rank] * self.world_size
+
+    mock_metadata = MockDPMetadata(world_size, num_tokens_per_rank)
+    mock_attn_metadata = PlaceholderAttnMetadata()
+    mock_attn_metadata.dp_metadata = mock_metadata
+
+    # Create VllmConfig for forward context with proper parallel config
+    from vllm.config.parallel import ParallelConfig
+    from vllm.forward_context import get_forward_context
+
+    vllm_config = VllmConfig()
+    vllm_config.parallel_config = ParallelConfig(
+        data_parallel_size=world_size, is_moe_model=True, data_parallel_rank=rank
+    )
+
+    # Create num_tokens_across_dp for all ranks
+    num_tokens_across_dp = torch.tensor(
+        [num_tokens_per_rank] * world_size, dtype=torch.int, device="cpu"
+    )
+
+    with set_forward_context(
+        mock_attn_metadata,
+        vllm_config,
+        num_tokens=num_tokens_per_rank,
+        num_tokens_across_dp=num_tokens_across_dp,
+    ):
+        # Initialize reference manager (AgRs - All-Gather/Reduce-Scatter)
+        reference_manager = AgRsAll2AllManager(cpu_group)
+
+        # Get dp_metadata and use sp_local_sizes context manager
+        dp_metadata = get_forward_context().dp_metadata
+        with dp_metadata.sp_local_sizes(sequence_parallel_size=1):
+            # Test 1: dispatch_router_logits
+            print(f"[Rank {rank}] Testing dispatch_router_logits")
+            ref_hidden, ref_router = reference_manager.dispatch_router_logits(
+                hidden_states.clone(), router_logits.clone(), is_sequence_parallel=True
+            )
+
+            # Test 2: dispatch
+            print(f"[Rank {rank}] Testing dispatch")
+            ref_hidden2, ref_weights, ref_ids = reference_manager.dispatch(
+                hidden_states.clone(),
+                topk_weights.clone(),
+                topk_ids.clone(),
+                is_sequence_parallel=True,
+            )
+
+            # Test 3: combine
+            print(f"[Rank {rank}] Testing combine")
+            # Create output tensor for combine (simulating expert outputs)
+            expert_output = torch.randn(
+                world_size * num_tokens_per_rank,
+                hidden_size,
+                device=device,
+                dtype=torch.float16,
+            )
+            ref_combined = reference_manager.combine(
+                expert_output.clone(), is_sequence_parallel=True
+            )
+
+            torch.distributed.barrier()
+
+            print(f"[Rank {rank}] ✓ Data communication validated successfully")
+            print(
+                f"[Rank {rank}]   - Dispatched hidden states shape: {ref_hidden.shape}"
+            )
+            print(f"[Rank {rank}]   - Combined output shape: {ref_combined.shape}")
+
+            torch.distributed.barrier()
+
+
+@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Need at least 2 GPUs")
+@pytest.mark.parametrize("world_size", [2])
+def test_alltoall_data_communication(world_size: int):
+    """
+    Test that all2all backends correctly communicate data across ranks.
+
+    This test validates data communication by:
+    1. Creating test tensors on each rank
+    2. Running dispatch and combine operations
+    3. Comparing results across different backends (AgRs, Naive)
+    4. Ensuring data is correctly exchanged between ranks
+    """
+    import torch.multiprocessing as mp
+
+    mp.set_start_method("spawn", force=True)
+
+    port = "12358"
+
+    # Launch multiple processes
+    processes = []
+    for rank in range(world_size):
+        p = mp.Process(
+            target=run_multi_gpu_test,
+            args=(rank, world_size, port, data_communication_worker),
+        )
+        p.start()
+        processes.append(p)
+
+    # Wait for all processes
+    for p in processes:
+        p.join()
+        assert p.exitcode == 0, f"Process failed with exit code {p.exitcode}"
+
+
+def flashinfer_data_communication_worker(rank: int, world_size: int):
+    """
+    Worker function for testing All2All data communication with value validation.
+
+    This test validates that AgRsAll2AllManager correctly communicates data
+    across ranks by checking that dispatched tensors contain contributions from
+    all ranks, not just shape validation.
+    """
+    from vllm.config.vllm import VllmConfig
+    from vllm.distributed.device_communicators.all2all import (
+        AgRsAll2AllManager,
+    )
+    from vllm.forward_context import set_forward_context
+
+    # Get CPU group
+    cpu_group = get_ep_group().cpu_group
+
+    print(f"[Rank {rank}] Testing All2All data communication with value validation")
+
+    # Test dimensions
+    hidden_size = 256
+    num_tokens_per_rank = 64
+    num_experts_per_token = 2
+    num_global_experts = world_size * 8  # 8 experts per rank
+
+    # Create test input data with DETERMINISTIC VALUES (unique per rank)
+    # This allows us to verify that data from all ranks is present after dispatch
+    device = torch.device(f"cuda:{rank}")
+
+    # Each rank uses a unique value: rank + 1
+    # This makes it easy to verify data is correctly gathered from all ranks
+    hidden_states = torch.full(
+        (num_tokens_per_rank, hidden_size),
+        float(rank + 1),
+        device=device,
+        dtype=torch.float16,
+    )
+    topk_weights = torch.full(
+        (num_tokens_per_rank, num_experts_per_token),
+        float(rank + 1) * 10,
+        device=device,
+        dtype=torch.float16,
+    )
+    topk_ids = torch.full(
+        (num_tokens_per_rank, num_experts_per_token),
+        rank,
+        device=device,
+        dtype=torch.long,
+    )
+    router_logits = torch.full(
+        (num_tokens_per_rank, num_global_experts),
+        float(rank + 1) * 100,
+        device=device,
+        dtype=torch.float16,
+    )
+
+    # Create mock forward context with dp_metadata
+    class MockDPMetadata:
+        def __init__(self, world_size, num_tokens_per_rank):
+            self.world_size = world_size
+            self.num_tokens_per_rank = num_tokens_per_rank
+
+        def cu_tokens_across_sp(self, sp_size):
+            cu_tokens = torch.tensor(
+                [i * self.num_tokens_per_rank for i in range(1, self.world_size + 1)],
+                dtype=torch.int32,
+            )
+            return cu_tokens
+
+        def get_chunk_sizes_across_dp_rank(self):
+            return [self.num_tokens_per_rank] * self.world_size
+
+    mock_metadata = MockDPMetadata(world_size, num_tokens_per_rank)
+    mock_attn_metadata = PlaceholderAttnMetadata()
+    mock_attn_metadata.dp_metadata = mock_metadata
+
+    # Create VllmConfig for forward context with proper parallel config
+    from vllm.config.parallel import ParallelConfig
+    from vllm.forward_context import get_forward_context
+
+    vllm_config = VllmConfig()
+    vllm_config.parallel_config = ParallelConfig(
+        data_parallel_size=world_size, is_moe_model=True, data_parallel_rank=rank
+    )
+
+    # Create num_tokens_across_dp for all ranks
+    num_tokens_across_dp = torch.tensor(
+        [num_tokens_per_rank] * world_size, dtype=torch.int, device="cpu"
+    )
+
+    with set_forward_context(
+        mock_attn_metadata,
+        vllm_config,
+        num_tokens=num_tokens_per_rank,
+        num_tokens_across_dp=num_tokens_across_dp,
+    ):
+        # Initialize All2All manager
+        manager = AgRsAll2AllManager(cpu_group)
+
+        # Get dp_metadata and use sp_local_sizes context manager
+        dp_metadata = get_forward_context().dp_metadata
+        with dp_metadata.sp_local_sizes(sequence_parallel_size=1):
+            expected_total_tokens = world_size * num_tokens_per_rank
+
+            # Test 1: dispatch_router_logits with value validation
+            print(f"[Rank {rank}] Testing dispatch_router_logits with value validation")
+            dispatched_hidden, dispatched_router = manager.dispatch_router_logits(
+                hidden_states.clone(), router_logits.clone(), is_sequence_parallel=True
+            )
+
+            # Validate shapes
+            assert dispatched_hidden.shape == (expected_total_tokens, hidden_size), (
+                f"[Rank {rank}] Unexpected hidden shape: {dispatched_hidden.shape}"
+            )
+            assert dispatched_router.shape == (expected_total_tokens, num_global_experts), (
+                f"[Rank {rank}] Unexpected router shape: {dispatched_router.shape}"
+            )
+
+            # Validate VALUES: verify data from all ranks is present
+            # Each rank's data should have its unique value (rank + 1)
+            for r in range(world_size):
+                start_idx = r * num_tokens_per_rank
+                end_idx = (r + 1) * num_tokens_per_rank
+                rank_hidden = dispatched_hidden[start_idx:end_idx, :]
+                rank_router = dispatched_router[start_idx:end_idx, :]
+
+                expected_hidden_val = float(r + 1)
+                expected_router_val = float(r + 1) * 100
+
+                actual_hidden_mean = rank_hidden.mean().item()
+                actual_router_mean = rank_router.mean().item()
+
+                assert abs(actual_hidden_mean - expected_hidden_val) < 0.1, (
+                    f"[Rank {rank}] Hidden states: expected rank {r} data to have value "
+                    f"{expected_hidden_val}, but got {actual_hidden_mean}"
+                )
+                assert abs(actual_router_mean - expected_router_val) < 10, (
+                    f"[Rank {rank}] Router logits: expected rank {r} data to have value "
+                    f"{expected_router_val}, but got {actual_router_mean}"
+                )
+
+            print(f"[Rank {rank}]   ✓ dispatch_router_logits: all rank data verified")
+
+            # Test 2: dispatch with value validation
+            print(f"[Rank {rank}] Testing dispatch with value validation")
+            dispatched_hidden2, dispatched_weights, dispatched_ids = manager.dispatch(
+                hidden_states.clone(),
+                topk_weights.clone(),
+                topk_ids.clone(),
+                is_sequence_parallel=True,
+            )
+
+            # Validate shapes
+            assert dispatched_hidden2.shape == (expected_total_tokens, hidden_size)
+            assert dispatched_weights.shape == (expected_total_tokens, num_experts_per_token)
+            assert dispatched_ids.shape == (expected_total_tokens, num_experts_per_token)
+
+            # Validate VALUES: verify data from all ranks
+            for r in range(world_size):
+                start_idx = r * num_tokens_per_rank
+                end_idx = (r + 1) * num_tokens_per_rank
+                rank_weights = dispatched_weights[start_idx:end_idx, :]
+                rank_ids = dispatched_ids[start_idx:end_idx, :]
+
+                expected_weight_val = float(r + 1) * 10
+                expected_id_val = r
+
+                actual_weight_mean = rank_weights.mean().item()
+                actual_id_val = rank_ids[0, 0].item()  # All IDs should be the same
+
+                assert abs(actual_weight_mean - expected_weight_val) < 1.0, (
+                    f"[Rank {rank}] Weights: expected rank {r} data to have value "
+                    f"{expected_weight_val}, but got {actual_weight_mean}"
+                )
+                assert actual_id_val == expected_id_val, (
+                    f"[Rank {rank}] IDs: expected rank {r} data to have value "
+                    f"{expected_id_val}, but got {actual_id_val}"
+                )
+
+            print(f"[Rank {rank}]   ✓ dispatch: all rank data verified")
+
+            # Test 3: combine with deterministic pattern
+            print(f"[Rank {rank}] Testing combine with value validation")
+            # Create expert output where each token position has a unique value
+            expert_output = torch.zeros(
+                expected_total_tokens, hidden_size, device=device, dtype=torch.float16
+            )
+            for i in range(expected_total_tokens):
+                expert_output[i, :] = float(i)
+
+            combined = manager.combine(expert_output, is_sequence_parallel=True)
+
+            # Validate shape
+            assert combined.shape == (num_tokens_per_rank, hidden_size), (
+                f"[Rank {rank}] Unexpected combined shape: {combined.shape}"
+            )
+
+            # Validate VALUES: after reduce-scatter, each rank gets its portion
+            # Rank 0 gets tokens [0, num_tokens_per_rank)
+            # Rank 1 gets tokens [num_tokens_per_rank, 2*num_tokens_per_rank)
+            expected_start_token = rank * num_tokens_per_rank
+            for i in range(num_tokens_per_rank):
+                # Due to all_reduce in reduce_scatter, values are summed across ranks
+                expected_value = float(expected_start_token + i) * world_size
+                actual_mean = combined[i, :].mean().item()
+
+                assert abs(actual_mean - expected_value) < 1.0, (
+                    f"[Rank {rank}] Token {i}: expected {expected_value}, got {actual_mean}"
+                )
+
+            print(f"[Rank {rank}]   ✓ combine: values correctly reduced")
+
+            torch.distributed.barrier()
+
+            print(f"[Rank {rank}] ✓ All2All data communication validation passed")
+            print(f"[Rank {rank}]   - Verified data from all {world_size} ranks is present")
+            print(f"[Rank {rank}]   - Verified dispatch gathers data correctly")
+            print(f"[Rank {rank}]   - Verified combine reduces data correctly")
+
+            torch.distributed.barrier()
+
+
+@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Need at least 2 GPUs")
+@pytest.mark.parametrize("world_size", [2])
+def test_flashinfer_alltoall_data_communication(world_size: int):
+    """
+    Test All2All data communication with value validation.
+
+    This test validates that AgRsAll2AllManager correctly communicates data
+    across ranks by using deterministic input values (each rank has unique values)
+    and verifying that:
+
+    1. dispatch_router_logits: gathered tensors contain data from ALL ranks
+    2. dispatch: gathered weights and IDs contain data from ALL ranks
+    3. combine: reduce-scatter correctly reduces data back to each rank
+
+    This goes beyond shape validation to ensure actual data values are correctly
+    communicated, addressing the requirement to "check the values match as well".
+    """
+    import torch.multiprocessing as mp
+
+    mp.set_start_method("spawn", force=True)
+
+    port = "12359"
+
+    # Launch multiple processes
+    processes = []
+    for rank in range(world_size):
+        p = mp.Process(
+            target=run_multi_gpu_test,
+            args=(rank, world_size, port, flashinfer_data_communication_worker),
+        )
+        p.start()
+        processes.append(p)
+
+    # Wait for all processes
+    for p in processes:
+        p.join()
+        assert p.exitcode == 0, f"Process failed with exit code {p.exitcode}"
+
+
+def deterministic_data_validation_worker(rank: int, world_size: int):
+    """
+    Worker function for validating exact data correctness with deterministic patterns.
+
+    This test creates deterministic data patterns where each rank has unique
+    values, then validates that dispatch correctly gathers data from all ranks
+    and combine correctly reduces it back.
+    """
+    from vllm.config.vllm import VllmConfig
+    from vllm.distributed.device_communicators.all2all import (
+        AgRsAll2AllManager,
+    )
+    from vllm.forward_context import set_forward_context
+
+    cpu_group = get_ep_group().cpu_group
+    device = torch.device(f"cuda:{rank}")
+
+    # Test dimensions
+    hidden_size = 64  # Smaller for easier debugging
+    num_tokens_per_rank = 16
+    num_experts_per_token = 2
+    num_global_experts = world_size * 4
+
+    # Create deterministic data: each rank has values = rank + 1
+    # This makes it easy to verify data is correctly communicated
+    hidden_states = torch.full(
+        (num_tokens_per_rank, hidden_size),
+        float(rank + 1),
+        device=device,
+        dtype=torch.float32,
+    )
+    router_logits = torch.full(
+        (num_tokens_per_rank, num_global_experts),
+        float(rank + 1) * 10,
+        device=device,
+        dtype=torch.float32,
+    )
+    topk_weights = torch.full(
+        (num_tokens_per_rank, num_experts_per_token),
+        float(rank + 1) * 100,
+        device=device,
+        dtype=torch.float32,
+    )
+    topk_ids = torch.full(
+        (num_tokens_per_rank, num_experts_per_token),
+        rank,
+        device=device,
+        dtype=torch.long,
+    )
+
+    # Create mock forward context
+    class MockDPMetadata:
+        def __init__(self, world_size, num_tokens_per_rank):
+            self.world_size = world_size
+            self.num_tokens_per_rank = num_tokens_per_rank
+
+        def cu_tokens_across_sp(self, sp_size):
+            return torch.tensor(
+                [i * self.num_tokens_per_rank for i in range(1, self.world_size + 1)],
+                dtype=torch.int32,
+            )
+
+        def get_chunk_sizes_across_dp_rank(self):
+            return [self.num_tokens_per_rank] * self.world_size
+
+    mock_metadata = MockDPMetadata(world_size, num_tokens_per_rank)
+    mock_attn_metadata = PlaceholderAttnMetadata()
+    mock_attn_metadata.dp_metadata = mock_metadata
+
+    # Create VllmConfig for forward context with proper parallel config
+    from vllm.config.parallel import ParallelConfig
+    from vllm.forward_context import get_forward_context
+
+    vllm_config = VllmConfig()
+    vllm_config.parallel_config = ParallelConfig(
+        data_parallel_size=world_size, is_moe_model=True, data_parallel_rank=rank
+    )
+
+    # Create num_tokens_across_dp for all ranks
+    num_tokens_across_dp = torch.tensor(
+        [num_tokens_per_rank] * world_size, dtype=torch.int, device="cpu"
+    )
+
+    with set_forward_context(
+        mock_attn_metadata,
+        vllm_config,
+        num_tokens=num_tokens_per_rank,
+        num_tokens_across_dp=num_tokens_across_dp,
+    ):
+        # Initialize manager
+        manager = AgRsAll2AllManager(cpu_group)
+
+        # Get dp_metadata and use sp_local_sizes context manager
+        dp_metadata = get_forward_context().dp_metadata
+        with dp_metadata.sp_local_sizes(sequence_parallel_size=1):
+            # Test dispatch_router_logits
+            print(f"[Rank {rank}] Testing deterministic dispatch_router_logits")
+            dispatched_hidden, dispatched_router = manager.dispatch_router_logits(
+                hidden_states.clone(),
+                router_logits.clone(),
+                is_sequence_parallel=True,
+            )
+
+            # Validate dispatched data contains contributions from all ranks
+            expected_total_tokens = world_size * num_tokens_per_rank
+            assert dispatched_hidden.shape[0] == expected_total_tokens, (
+                f"[Rank {rank}] Expected {expected_total_tokens} tokens, "
+                f"got {dispatched_hidden.shape[0]}"
+            )
+
+            # Verify that dispatched data contains values from all ranks
+            # After all_gatherv, we should have concatenated data from all ranks
+            for r in range(world_size):
+                start_idx = r * num_tokens_per_rank
+                end_idx = (r + 1) * num_tokens_per_rank
+                rank_data = dispatched_hidden[start_idx:end_idx, :]
+
+                # Each rank's data should have value = r + 1
+                expected_value = float(r + 1)
+                actual_mean = rank_data.mean().item()
+
+                assert abs(actual_mean - expected_value) < 1e-4, (
+                    f"[Rank {rank}] Expected rank {r} data to have value "
+                    f"{expected_value}, but got {actual_mean}"
+                )
+
+            print(f"[Rank {rank}] ✓ Dispatch validation passed - all rank data present")
+
+            # Test combine with deterministic pattern
+            # Create expert output where each token has value = token_index
+            expert_output = torch.zeros(
+                expected_total_tokens,
+                hidden_size,
+                device=device,
+                dtype=torch.float32,
+            )
+            for i in range(expected_total_tokens):
+                expert_output[i, :] = float(i)
+
+            combined = manager.combine(expert_output, is_sequence_parallel=True)
+
+            # After reduce_scatterv, each rank should get its portion
+            # Rank 0 gets tokens [0, num_tokens_per_rank)
+            # Rank 1 gets tokens [num_tokens_per_rank, 2*num_tokens_per_rank)
+            # etc.
+            expected_start_token = rank * num_tokens_per_rank
+            for i in range(num_tokens_per_rank):
+                expected_value = (
+                    float(expected_start_token + i) * world_size
+                )  # Due to all_reduce
+                actual_mean = combined[i, :].mean().item()
+
+                assert abs(actual_mean - expected_value) < 1e-3, (
+                    f"[Rank {rank}] Token {i}: expected {expected_value}, "
+                    f"got {actual_mean}"
+                )
+
+            print(f"[Rank {rank}] ✓ Combine validation passed - correct data reduction")
+
+            # Test dispatch with topk
+            print(f"[Rank {rank}] Testing deterministic dispatch")
+            dispatched_hidden2, dispatched_weights, dispatched_ids = manager.dispatch(
+                hidden_states.clone(),
+                topk_weights.clone(),
+                topk_ids.clone(),
+                is_sequence_parallel=True,
+            )
+
+            # Verify shapes
+            assert dispatched_hidden2.shape == (expected_total_tokens, hidden_size)
+            assert dispatched_weights.shape == (
+                expected_total_tokens,
+                num_experts_per_token,
+            )
+            assert dispatched_ids.shape == (
+                expected_total_tokens,
+                num_experts_per_token,
+            )
+
+            # Verify weights contain data from all ranks
+            for r in range(world_size):
+                start_idx = r * num_tokens_per_rank
+                end_idx = (r + 1) * num_tokens_per_rank
+                rank_weights = dispatched_weights[start_idx:end_idx, :]
+
+                expected_value = float(r + 1) * 100
+                actual_mean = rank_weights.mean().item()
+
+                assert abs(actual_mean - expected_value) < 1e-3, (
+                    f"[Rank {rank}] Expected rank {r} weights to be {expected_value}, "
+                    f"got {actual_mean}"
+                )
+
+            print(f"[Rank {rank}] ✓ All deterministic data validation passed")
+
+            torch.distributed.barrier()
+
+
+@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Need at least 2 GPUs")
+@pytest.mark.parametrize("world_size", [2])
+def test_alltoall_deterministic_data_validation(world_size: int):
+    """
+    Test data correctness with deterministic patterns.
+
+    This test validates that:
+    1. Dispatch correctly gathers data from all ranks (all_gatherv semantics)
+    2. Combine correctly reduces data back to each rank (reduce_scatterv semantics)
+    3. Actual data values are preserved and correctly communicated
+
+    Uses deterministic patterns where each rank has unique, verifiable values.
+    """
+    import torch.multiprocessing as mp
+
+    mp.set_start_method("spawn", force=True)
+
+    port = "12360"
+
+    processes = []
+    for rank in range(world_size):
+        p = mp.Process(
+            target=run_multi_gpu_test,
+            args=(rank, world_size, port, deterministic_data_validation_worker),
+        )
+        p.start()
+        processes.append(p)
+
+    for p in processes:
+        p.join()
+        assert p.exitcode == 0, f"Process failed with exit code {p.exitcode}"
+
+
+def test_custom_communicator():
+    """Test CustomCommunicator wrapper for FlashInfer."""
+    if not has_flashinfer_all2all():
+        pytest.skip("FlashInfer alltoall not available")
+
+    from vllm.distributed.device_communicators.mnnvl_compat import (
+        CustomCommunicator,
+    )
+
+    class MockGroup:
+        def rank(self):
+            return 0
+
+        def size(self):
+            return 2
+
+    mock_group = MockGroup()
+    comm = CustomCommunicator(mock_group)
+
+    # Test basic methods
+    assert comm.Get_rank() == 0
+    assert comm.Get_size() == 2
+
+    # Test unimplemented methods raise NotImplementedError
+    with pytest.raises(NotImplementedError):
+        comm.bcast(None, 0)
+
+    with pytest.raises(NotImplementedError):
+        comm.barrier()
+
+    # Test Split returns self (as per implementation)
+    split_comm = comm.Split(0, 0)
+    assert split_comm is comm
+
+
+if __name__ == "__main__":
+    # Check SYS_PTRACE capability first
+    print("=" * 70)
+    print("MNNVL AllToAll Test Configuration")
+    print("=" * 70)
+
+    print(f"\nGPUs available: {torch.cuda.device_count()}")
+    print(f"FlashInfer AllToAll available: {has_flashinfer_all2all()}")
+
+    if has_sys_ptrace_capability():
+        print("✓ SYS_PTRACE capability: DETECTED")
+        print("  Multi-GPU tests will run")
+    else:
+        print("⚠ SYS_PTRACE capability: NOT DETECTED")
+        print("  Multi-GPU tests will be skipped")
+        print("  To enable: docker run --cap-add=SYS_PTRACE ...")
+
+    print("\n" + "=" * 70)
+    print("Running Standalone Tests")
+    print("=" * 70 + "\n")
+
+    # Run basic import test
+    test_flashinfer_all2all_import()
+    print("✓ Import test passed")
+
+    # Run communicator test
+    test_custom_communicator()
+    print("✓ Custom communicator test passed")
+
+    print("\n" + "=" * 70)
+    print("All standalone tests passed!")
+    print("=" * 70)
+    print("\nTo run full multi-GPU test suite:")
+    print("  pytest tests/distributed/test_mnnvl_alltoall.py -v")