[DSV4] Cherry pick missing commits from deepseek_v4 branch and enhance tests (#24793)

Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com> Co-authored-by: yueming-yuan <yym022502@gmail.com>
2026-05-09 04:15:37 -07:00
parent 4b23f6bdc5
commit ef5e9f8aba
15 changed files with 481 additions and 87 deletions
@@ -1454,7 +1454,7 @@ jobs:
      - name: Install dependencies
        timeout-minutes: 30
        run: |
-          CUSTOM_BUILD_SGL_KERNEL=${{needs.check-changes.outputs.sgl_kernel}} bash scripts/ci/cuda/ci_install_flash_mla.sh
+          CUSTOM_BUILD_SGL_KERNEL=${{needs.check-changes.outputs.sgl_kernel}} bash scripts/ci/cuda/ci_install_dsv4_dep.sh

      - name: Run test
        timeout-minutes: 30
@@ -1506,7 +1506,7 @@ jobs:
      - name: Install dependencies
        timeout-minutes: 30
        run: |
-          CUSTOM_BUILD_SGL_KERNEL=${{needs.check-changes.outputs.sgl_kernel}} bash scripts/ci/cuda/ci_install_flash_mla.sh
+          CUSTOM_BUILD_SGL_KERNEL=${{needs.check-changes.outputs.sgl_kernel}} bash scripts/ci/cuda/ci_install_dsv4_dep.sh

      - name: Run test
        timeout-minutes: 30
@@ -378,7 +378,6 @@ export const DeepSeekV4Deployment = () => {
          "SGLANG_OPT_USE_CUSTOM_ALL_REDUCE_V2=1",
          "SGLANG_OPT_SWA_EVICT_DROP_PAGE_MARGIN=1",
          "SGLANG_OPT_USE_DEEPGEMM_MEGA_MOE=0",
-          "SGLANG_OPT_FIX_HASH_MEGA_MOE=0",
          "SGLANG_OPT_USE_FAST_MASK_EP=1",
          "SGLANG_OPT_FIX_MEGA_MOE_MEMORY=1",
          "SGLANG_OPT_DEEPGEMM_MEGA_MOE_NUM_MAX_TOKENS_PER_RANK=4096",
@@ -412,7 +411,6 @@ export const DeepSeekV4Deployment = () => {
          "NVSHMEM_DISABLE_IB=1",
          "SGLANG_OPT_SWA_RELEASE_LEAF_LOCK_AFTER_WINDOW=1",
          "SGLANG_OPT_USE_DEEPGEMM_MEGA_MOE=1",
-          "SGLANG_OPT_FIX_HASH_MEGA_MOE=1",
          "SGLANG_OPT_DEEPGEMM_MEGA_MOE_NUM_MAX_TOKENS_PER_RANK=8320",
        );
      } else {
@@ -926,7 +924,6 @@ python3 -m sglang_router.launch_router \\

 # And set these env vars:
 SGLANG_OPT_USE_DEEPGEMM_MEGA_MOE=1
-SGLANG_OPT_FIX_HASH_MEGA_MOE=1
 SGLANG_OPT_FIX_MEGA_MOE_MEMORY=1
 SGLANG_OPT_FIX_NEXTN_MEGA_MOE=1
 SGLANG_OPT_DEEPGEMM_MEGA_MOE_NUM_MAX_TOKENS_PER_RANK=8320
@@ -633,13 +633,16 @@ class ChatCompletionRequest(BaseModel):
    return_hidden_states: bool = False
    return_routed_experts: bool = False
    return_cached_tokens_details: bool = False
-    reasoning_effort: Optional[Literal["none", "low", "medium", "high"]] = Field(
+    reasoning_effort: Optional[Literal["none", "low", "medium", "high", "max"]] = Field(
        default=None,
        description="Constrains effort on reasoning for reasoning models. "
        "'none' disables reasoning entirely, 'low' is the least effort, 'high' is the most effort. "
        "Reducing reasoning effort can result in faster responses and fewer tokens used on reasoning "
        "in a response. 'none' defaults thinking and enable_thinking to false in "
-        "chat_template_kwargs (unless explicitly overridden). Not supported in the harmony path.",
+        "chat_template_kwargs (unless explicitly overridden). Not supported in the harmony path."
+        "'max' is an sglang extension to the OpenAI schema for "
+        "models that expose a maximum-effort tier above 'high'; models that don't "
+        "support it treat it the same as 'high'.",
    )
    task: Optional[
        Literal["action", "query", "authority", "domain", "title", "read_url"]
@@ -81,8 +81,13 @@ class DeepSeekV32Detector(BaseFormatDetector):
        self.function_calls_regex = (
            r"<｜DSML｜function_calls>(.*?)</｜DSML｜function_calls>"
        )
+        # Long-form `<｜DSML｜invoke name="x">...</｜DSML｜invoke>` and the
+        # self-closing `<｜DSML｜invoke name="x"/>` shape V4 emits for zero-arg
+        # tools. The `end` group is empty when the closer hasn't streamed in.
        self.invoke_regex = (
-            r'<｜DSML｜invoke\s+name="([^"]+)"\s*>(.*?)(</｜DSML｜invoke>|$)'
+            r'<｜DSML｜invoke\s+name="(?P<name>[^"]+)"\s*'
+            r"(?:(?P<self_close>/>)"
+            r"|>(?P<body>.*?)(?P<end>(?:</｜DSML｜invoke>|$)))"
        )
        self.prefix_parameter_end_call = ["</", "｜DSML｜", "parameter"]
        self.prefix_invoke_end_call = ["</", "｜DSML｜", "inv", "oke"]
@@ -92,6 +97,20 @@ class DeepSeekV32Detector(BaseFormatDetector):
        """Check if the text contains a deepseek v32 format tool call."""
        return self.bot_token in text or "<｜DSML｜invoke" in text

+    @staticmethod
+    def _unpack_invoke_match(m: "re.Match[str]") -> tuple[str, str, bool]:
+        """Returns (name, body, is_complete) for an invoke_regex match.
+
+        Self-closing invokes have empty body and are always complete.
+        Long-form bodies are always strings (possibly empty); they're
+        incomplete when matched against `$` because the closing tag
+        hasn't streamed in yet.
+        """
+        name = m.group("name").strip()
+        if m.group("self_close"):
+            return name, "", True
+        return name, m.group("body"), bool(m.group("end"))
+
    def _parse_parameters_from_xml(
        self, invoke_content: str, allow_partial: bool = False
    ) -> str:
@@ -192,12 +211,10 @@ class DeepSeekV32Detector(BaseFormatDetector):
            function_calls_content = function_calls_match.group(1)

            # Find all invoke blocks
-            invoke_matches = re.findall(
+            for invoke_match in re.finditer(
                self.invoke_regex, function_calls_content, re.DOTALL
-            )
-
-            for func_name, invoke_content, _ in invoke_matches:
-                # Parse parameters from XML format
+            ):
+                func_name, invoke_content, _ = self._unpack_invoke_match(invoke_match)
                func_args = self._parse_parameters_from_xml(invoke_content)
                # construct match_result for parse_base_json
                match_result = {"name": func_name, "parameters": json.loads(func_args)}
@@ -254,10 +271,9 @@ class DeepSeekV32Detector(BaseFormatDetector):
                if not invoke_match:
                    break

-                func_name = invoke_match.group(1).strip()
-                invoke_content = invoke_match.group(2)
-                # group(3) is either "</｜DSML｜invoke>" (complete) or "" (incomplete, matched with $)
-                is_tool_end = bool(invoke_match.group(3))
+                func_name, invoke_content, is_tool_end = self._unpack_invoke_match(
+                    invoke_match
+                )

                # Initialize state if this is the first tool call
                if self.current_tool_id == -1:
@@ -528,6 +528,9 @@ class DefaultModelLoader(BaseModelLoader):
            weight_loader_disable_mmap = server_args.weight_loader_disable_mmap
            weight_loader_prefetch = server_args.weight_loader_prefetch_checkpoints
            prefetch_num_threads = server_args.weight_loader_prefetch_num_threads
+            weight_loader_drop_cache_after_load = (
+                server_args.weight_loader_drop_cache_after_load
+            )

            if self.load_config.load_format == LoadFormat.FASTSAFETENSORS:
                weights_iterator = fastsafetensors_weights_iterator(
@@ -542,6 +545,7 @@ class DefaultModelLoader(BaseModelLoader):
                    disable_mmap=weight_loader_disable_mmap,
                    prefetch=weight_loader_prefetch,
                    prefetch_num_threads=prefetch_num_threads,
+                    drop_cache_after_load=weight_loader_drop_cache_after_load,
                )
            else:
                weights_iterator = safetensors_weights_iterator(
@@ -549,6 +553,7 @@ class DefaultModelLoader(BaseModelLoader):
                    disable_mmap=weight_loader_disable_mmap,
                    prefetch=weight_loader_prefetch,
                    prefetch_num_threads=prefetch_num_threads,
+                    drop_cache_after_load=weight_loader_drop_cache_after_load,
                )

        else:
@@ -875,11 +875,30 @@ def _prefetch_all_checkpoints(
    threading.Thread(target=_run_prefetch, daemon=True).start()


+def _drop_file_cache_after_load(path: str) -> None:
+    """Release of checkpoint pages after weights have been copied out. Used to avoid CPU OOM in RL."""
+    posix_fadvise = getattr(os, "posix_fadvise", None)
+    dontneed = getattr(os, "POSIX_FADV_DONTNEED", None)
+    if posix_fadvise is None or dontneed is None:
+        return
+
+    fd = None
+    try:
+        fd = os.open(path, os.O_RDONLY)
+        posix_fadvise(fd, 0, 0, dontneed)
+    except OSError as e:
+        logger.debug("Failed to drop file cache for %s: %s", path, e)
+    finally:
+        if fd is not None:
+            os.close(fd)
+
+
 def safetensors_weights_iterator(
    hf_weights_files: List[str],
    disable_mmap: bool = False,
    prefetch: bool = False,
    prefetch_num_threads: int = 4,
+    drop_cache_after_load: bool = False,
 ) -> Generator[Tuple[str, torch.Tensor], None, None]:
    """Iterate over the weights in the model safetensor files."""
    enable_tqdm = (
@@ -907,6 +926,8 @@ def safetensors_weights_iterator(
            with safetensors.safe_open(st_file, framework="pt", device="cpu") as f:
                for name in f.keys():
                    yield name, f.get_tensor(name)
+        if drop_cache_after_load:
+            _drop_file_cache_after_load(st_file)


 def fastsafetensors_weights_iterator(
@@ -968,6 +989,7 @@ def multi_thread_safetensors_weights_iterator(
    hf_weights_files: List[str],
    max_workers: int,
    disable_mmap: bool = False,
+    drop_cache_after_load: bool = False,
 ) -> Generator[Tuple[str, torch.Tensor], None, None]:
    """Multi-Thread iterate over the weights in the model safetensor files."""
    enable_tqdm = (
@@ -977,8 +999,12 @@ def multi_thread_safetensors_weights_iterator(
    def _load_file(st_file: str):
        if disable_mmap:
            with open(st_file, "rb") as f:
-                return safetensors.torch.load(f.read())
-        return safetensors.torch.load_file(st_file, device="cpu")
+                result = safetensors.torch.load(f.read())
+        else:
+            with safetensors.safe_open(st_file, framework="pt", device="cpu") as f:
+                result = {k: f.get_tensor(k) for k in f.keys()}
+
+        return st_file, result

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(_load_file, st_file) for st_file in hf_weights_files]
@@ -995,9 +1021,12 @@ def multi_thread_safetensors_weights_iterator(
            futures_iter = concurrent.futures.as_completed(futures)

        for future in futures_iter:
-            state_dict = future.result()
+            st_file, state_dict = future.result()
            for name, param in state_dict.items():
                yield name, param
+            del state_dict
+            if drop_cache_after_load:
+                _drop_file_cache_after_load(st_file)


 def buffered_multi_thread_safetensors_weights_iterator(
@@ -1006,6 +1035,7 @@ def buffered_multi_thread_safetensors_weights_iterator(
    disable_mmap: bool = False,
    prefetch: bool = False,
    prefetch_num_threads: int = 4,
+    drop_cache_after_load: bool = False,
 ) -> Generator[Tuple[str, torch.Tensor], None, None]:
    """Multi-threaded safetensor loader with bounded memory via a sliding window.

@@ -791,6 +791,7 @@ class ServerArgs:
    weight_loader_disable_mmap: bool = False
    weight_loader_prefetch_checkpoints: bool = False
    weight_loader_prefetch_num_threads: int = 4
+    weight_loader_drop_cache_after_load: bool = False
    remote_instance_weight_loader_seed_instance_ip: Optional[str] = None
    remote_instance_weight_loader_seed_instance_service_port: Optional[int] = None
    remote_instance_weight_loader_send_weights_group_ports: Optional[List[int]] = None
@@ -6658,6 +6659,11 @@ class ServerArgs:
            default=ServerArgs.weight_loader_prefetch_num_threads,
            help="Number of threads per rank for checkpoint prefetching (default: 4).",
        )
+        parser.add_argument(
+            "--weight-loader-drop-cache-after-load",
+            action="store_true",
+            help="Call posix_fadvise(DONTNEED) on each safetensors shard after loading it.",
+        )
        parser.add_argument(
            "--remote-instance-weight-loader-seed-instance-ip",
            type=str,
@@ -0,0 +1,161 @@
+#!/bin/bash
+set -euxo pipefail
+
+source scripts/ci/cuda/ci_install_dependency.sh
+
+if [ -z "${PIP_CMD:-}" ]; then
+    echo "FATAL:PIP_CMD is unset after sourcing ci_install_dependency.sh"
+    exit 1
+fi
+
+export GDRCOPY_HOME=/usr/src/gdrdrv-2.5.1/
+export CUDA_HOME=/usr/local/cuda
+
+# Detect architecture
+ARCH=$(uname -m)
+if [ "$ARCH" != "x86_64" ] && [ "$ARCH" != "aarch64" ]; then
+    echo "Unsupported architecture: $ARCH"
+    exit 1
+fi
+
+###############################################################################
+# Install FlashMLA
+###############################################################################
+INSTALL_FLASH_MLA=1
+if [ "${FORCE_REBUILD_FLASH_MLA:-0}" = "1" ]; then
+    echo "FORCE_REBUILD_FLASH_MLA=1; uninstalling any cached flash_mla before rebuild."
+    ${PIP_UNINSTALL_CMD:-pip uninstall -y} flash_mla ${PIP_UNINSTALL_SUFFIX:-} || true
+elif python3 -c "import flash_mla" >/dev/null 2>&1; then
+    echo "flash_mla is already installed or importable. Skipping installation."
+    INSTALL_FLASH_MLA=0
+fi
+
+if [ "$INSTALL_FLASH_MLA" = "1" ]; then
+    # CUDA 13.0 puts CCCL headers under /usr/local/cuda/include/cccl/cuda but
+    # FlashMLA's build expects them at /usr/local/cuda/include/cuda. Symlink so
+    # the compiler finds them. Idempotent: skip if the link/dir already exists.
+    if [ ! -e /usr/local/cuda/include/cuda ] && [ -d /usr/local/cuda/include/cccl/cuda ]; then
+        ln -s /usr/local/cuda/include/cccl/cuda /usr/local/cuda/include/cuda
+    fi
+
+    FLASH_MLA_DIR=/root/.cache/flash-mla
+    rm -rf ${FLASH_MLA_DIR}
+    git clone https://github.com/deepseek-ai/FlashMLA.git ${FLASH_MLA_DIR}
+    pushd ${FLASH_MLA_DIR}
+    git submodule update --init --recursive
+    ${PIP_CMD:-pip} install --no-build-isolation -v . ${PIP_INSTALL_SUFFIX:-}
+    popd
+fi
+
+###############################################################################
+# Install DeepEP
+###############################################################################
+# Default to a forced rebuild so changes to TORCH_CUDA_ARCH_LIST or any other
+# build-time input don't silently reuse a cached deep_ep from a prior run.
+INSTALL_DEEPEP=1
+if [ "${FORCE_REBUILD_DEEPEP:-1}" = "1" ]; then
+    echo "FORCE_REBUILD_DEEPEP=1; uninstalling any cached deep_ep before rebuild."
+    ${PIP_UNINSTALL_CMD:-pip uninstall -y} deep_ep ${PIP_UNINSTALL_SUFFIX:-} || true
+elif python3 -c "import deep_ep" >/dev/null 2>&1; then
+    echo "deep_ep is already installed or importable. Skipping installation."
+    INSTALL_DEEPEP=0
+fi
+
+if [ "$INSTALL_DEEPEP" = "1" ]; then
+    # Install system dependencies
+    # Use fallback logic in case apt fails due to unrelated broken packages on the runner
+    DEEPEP_SYSTEM_DEPS="curl wget git sudo rdma-core infiniband-diags openssh-server perftest libibumad3 libibverbs-dev libibverbs1 ibverbs-providers ibverbs-utils libnl-3-200 libnl-route-3-200 librdmacm1 build-essential cmake"
+    apt-get install -y --no-install-recommends $DEEPEP_SYSTEM_DEPS || {
+        echo "Warning: apt-get install failed, checking if required packages are available..."
+        for pkg in $DEEPEP_SYSTEM_DEPS; do
+            if ! dpkg -l "$pkg" 2>/dev/null | grep -q "^ii"; then
+                echo "ERROR: Required package $pkg is not installed and apt-get failed"
+                exit 1
+            fi
+        done
+        echo "All required packages are already installed, continuing..."
+    }
+
+    # Install GDRCopy
+    rm -rf /opt/gdrcopy && mkdir -p /opt/gdrcopy
+    cd /opt/gdrcopy
+    git clone https://github.com/NVIDIA/gdrcopy.git .
+    git checkout v2.5.1
+    apt-get update || true  # May fail due to unrelated broken packages
+    GDRCOPY_DEPS_1="nvidia-dkms-580"
+    GDRCOPY_DEPS_2="build-essential devscripts debhelper fakeroot pkg-config dkms"
+    GDRCOPY_DEPS_3="check libsubunit0 libsubunit-dev python3-venv"
+    for deps_group in "$GDRCOPY_DEPS_1" "$GDRCOPY_DEPS_2" "$GDRCOPY_DEPS_3"; do
+        apt-get install -y --no-install-recommends $deps_group || {
+            echo "Warning: apt-get install failed for '$deps_group', checking if packages are available..."
+            for pkg in $deps_group; do
+                if ! dpkg -l "$pkg" 2>/dev/null | grep -q "^ii"; then
+                    echo "ERROR: Required package $pkg is not installed and apt-get failed"
+                    exit 1
+                fi
+            done
+            echo "All required packages from '$deps_group' are already installed, continuing..."
+        }
+    done
+    cd packages
+    CUDA=/usr/local/cuda ./build-deb-packages.sh
+    dpkg -i gdrdrv-dkms_*.deb
+    dpkg -i libgdrapi_*.deb
+    dpkg -i gdrcopy-tests_*.deb
+    dpkg -i gdrcopy_*.deb
+
+    # Set up library paths based on architecture
+    LIB_PATH="/usr/lib/$ARCH-linux-gnu"
+    if [ ! -e "$LIB_PATH/libmlx5.so" ]; then
+        ln -s $LIB_PATH/libmlx5.so.1 $LIB_PATH/libmlx5.so
+    fi
+    apt-get update || true
+    apt-get install -y --no-install-recommends libfabric-dev || {
+        if ! dpkg -l libfabric-dev 2>/dev/null | grep -q "^ii"; then
+            echo "ERROR: Required package libfabric-dev is not installed and apt-get failed"
+            exit 1
+        fi
+        echo "libfabric-dev is already installed, continuing..."
+    }
+
+    # Install DeepEP
+    DEEPEP_DIR=/root/.cache/deepep
+    rm -rf ${DEEPEP_DIR}
+    git clone https://github.com/deepseek-ai/DeepEP.git ${DEEPEP_DIR}
+    pushd ${DEEPEP_DIR}
+    git checkout 9af0e0d0e74f3577af1979c9b9e1ac2cad0104ee
+    popd
+
+    cd ${DEEPEP_DIR}
+    # CUDA 13.0 puts CCCL headers in /usr/local/cuda/include/cccl/ but nvshmem
+    # includes them as <cuda/__cccl_config> expecting /usr/local/cuda/include/cuda/.
+    # Add the cccl path to setup.py include_dirs so the compiler finds them.
+    NVCC_MAJOR=$(nvcc --version 2>/dev/null | grep -oP 'release \K[0-9]+' || echo "0")
+    if [ "$NVCC_MAJOR" = "13" ]; then
+        sed -i "/^    include_dirs = \['csrc\/'\]/a\    include_dirs.append('${CUDA_HOME:-/usr/local/cuda}/include/cccl')" setup.py
+    fi
+
+    # Build for both Hopper (sm_90) and Blackwell (sm_100) so the same wheel
+    # runs on H200 and B200 runners. Mirrors the CUDA-version-keyed list in
+    # docker/Dockerfile's DeepEP build stage.
+    if [ -n "${NVCC_VER:-}" ]; then
+        CUDA_VERSION="$NVCC_VER"
+    elif command -v nvcc >/dev/null 2>&1; then
+        CUDA_VERSION=$(nvcc --version | grep -oP 'release \K[0-9]+\.[0-9]+')
+    else
+        CUDA_VERSION=$(nvidia-smi | grep "CUDA Version" | head -n1 | awk '{print $9}' || true)
+    fi
+    if [ -z "${CUDA_VERSION:-}" ]; then
+        echo "FATAL: could not determine CUDA toolkit version (NVCC_VER unset, nvcc missing, nvidia-smi empty)"
+        exit 1
+    fi
+    if [ "$CUDA_VERSION" = "12.8" ]; then
+        CHOSEN_TORCH_CUDA_ARCH_LIST='9.0;10.0'
+    elif awk -v ver="$CUDA_VERSION" 'BEGIN {exit !(ver > 12.8)}'; then
+        # CUDA > 12.8 supports sm_103 (Blackwell)
+        CHOSEN_TORCH_CUDA_ARCH_LIST='9.0;10.0;10.3'
+    else
+        CHOSEN_TORCH_CUDA_ARCH_LIST='9.0'
+    fi
+    TORCH_CUDA_ARCH_LIST="${CHOSEN_TORCH_CUDA_ARCH_LIST}" python3 setup.py install
+fi
@@ -1,35 +0,0 @@
-#!/bin/bash
-set -euxo pipefail
-
-source scripts/ci/cuda/ci_install_dependency.sh
-
-if [ -z "${PIP_CMD:-}" ]; then
-    echo "FATAL:PIP_CMD is unset after sourcing ci_install_dependency.sh"
-    exit 1
-fi
-
-export CUDA_HOME=/usr/local/cuda
-
-if [ "${FORCE_REBUILD_FLASH_MLA:-0}" = "1" ]; then
-    echo "FORCE_REBUILD_FLASH_MLA=1; uninstalling any cached flash_mla before rebuild."
-    ${PIP_UNINSTALL_CMD:-pip uninstall -y} flash_mla ${PIP_UNINSTALL_SUFFIX:-} || true
-elif python3 -c "import flash_mla" >/dev/null 2>&1; then
-    echo "flash_mla is already installed or importable. Skipping installation."
-    exit 0
-fi
-
-# CUDA 13.0 puts CCCL headers under /usr/local/cuda/include/cccl/cuda but
-# FlashMLA's build expects them at /usr/local/cuda/include/cuda. Symlink so
-# the compiler finds them. Idempotent: skip if the link/dir already exists.
-if [ ! -e /usr/local/cuda/include/cuda ] && [ -d /usr/local/cuda/include/cccl/cuda ]; then
-    ln -s /usr/local/cuda/include/cccl/cuda /usr/local/cuda/include/cuda
-fi
-
-# Install FlashMLA
-FLASH_MLA_DIR=/root/.cache/flash-mla
-rm -rf ${FLASH_MLA_DIR}
-git clone https://github.com/deepseek-ai/FlashMLA.git ${FLASH_MLA_DIR}
-pushd ${FLASH_MLA_DIR}
-git submodule update --init --recursive
-${PIP_CMD:-pip} install --no-build-isolation -v . ${PIP_INSTALL_SUFFIX:-}
-popd
@@ -426,6 +426,8 @@ def handle_rerun_stage(
        "stage-c-test-4-gpu-gb200",
        "stage-c-test-deepep-4-gpu-h100",
        "stage-c-test-deepep-8-gpu-h200",
+        "stage-c-test-dsv4-4-gpu-b200",
+        "stage-c-test-dsv4-8-gpu-h200",
        "multimodal-gen-test-1-gpu",
        "multimodal-gen-test-2-gpu",
        "multimodal-gen-component-accuracy",
@@ -613,6 +615,8 @@ CUDA_SUITE_TO_RUNNER = {
    "stage-c-test-4-gpu-b200": "4-gpu-b200",
    "stage-c-test-deepep-4-gpu-h100": "4-gpu-h100",
    "stage-c-test-deepep-8-gpu-h200": "8-gpu-h200-deepep",
+    "stage-c-test-dsv4-4-gpu-b200": "4-gpu-b200",
+    "stage-c-test-dsv4-8-gpu-h200": "8-gpu-h200",
    # Nightly test suites (NVIDIA)
    "nightly-1-gpu": "1-gpu-h100",
    "nightly-4-gpu": "4-gpu-h100",
@@ -635,6 +639,8 @@ DEEPEP_SUITES = {
    "stage-c-test-8-gpu-h20",
    "stage-c-test-deepep-4-gpu-h100",
    "stage-c-test-deepep-8-gpu-h200",
+    "stage-c-test-dsv4-4-gpu-b200",
+    "stage-c-test-dsv4-8-gpu-h200",
 }


@@ -1,14 +1,10 @@
-"""B200 nightly CI: DeepSeek-V4-Flash FP4 (Balanced + MaxThroughput recipes).
+"""B200 per-commit CI: DeepSeek-V4-Flash FP4 (LowLatency recipe).

-Two server configurations exercise the DeepEP all-to-all + DP-attention path
-that the per-commit LowLatency test does not cover.
+Launches TP=4 with flashinfer_mxfp4 MoE runner + EAGLE speculative decoding.
+Runs 12 ServerSanity probes (correctness, streaming, concurrency, determinism)
+plus a GSM8K accuracy gate.

-  Balanced:       TP=4, DP=4, DeepEP, EAGLE (1 step)
-  MaxThroughput:  TP=4, DP=4, DeepEP, no speculation
-
-Each class inherits 12 ServerSanity probes plus a GSM8K accuracy gate.
-
-Registry: nightly-4-gpu-b200
+Registry: stage-c-test-dsv4-4-gpu-b200 (per-commit, 4x B200)
 """

 import unittest
@@ -25,7 +21,7 @@ from sglang.test.test_utils import (
    try_cached_model,
 )

-register_cuda_ci(est_time=3600, suite="nightly-4-gpu-b200", nightly=True)
+register_cuda_ci(est_time=1800, suite="stage-c-test-dsv4-4-gpu-b200")

 MODEL = "deepseek-ai/DeepSeek-V4-Flash"
 SERVER_LAUNCH_TIMEOUT = 3600
@@ -35,6 +31,14 @@ _DEEPEP_ENV = {
    "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": "1024",
 }

+_MEGAMOE_ENV = {
+    "SGLANG_OPT_USE_DEEPGEMM_MEGA_MOE": "1",
+    "SGLANG_OPT_FIX_MEGA_MOE_MEMORY": "1",
+    "SGLANG_OPT_FIX_NEXTN_MEGA_MOE": "1",
+    "SGLANG_OPT_DEEPGEMM_MEGA_MOE_NUM_MAX_TOKENS_PER_RANK": "4096",
+    "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": "0",
+}
+

 def _gsm8k_check(test_case):
    args = SimpleNamespace(
@@ -51,6 +55,46 @@ def _gsm8k_check(test_case):
    test_case.assertGreater(metrics["score"], 0.93)


+class TestDSV4FlashFP4B200(ServerSanityMixin, CustomTestCase):
+    """LowLatency recipe: TP=4, FP4 (mxfp4), EAGLE spec decoding."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.model = try_cached_model(MODEL)
+        cls.base_url = DEFAULT_URL_FOR_TEST
+        cls.process = popen_launch_server(
+            cls.model,
+            cls.base_url,
+            timeout=SERVER_LAUNCH_TIMEOUT,
+            other_args=[
+                "--trust-remote-code",
+                "--tp",
+                "4",
+                "--moe-runner-backend",
+                "flashinfer_mxfp4",
+                "--speculative-algorithm",
+                "EAGLE",
+                "--speculative-num-steps",
+                "3",
+                "--speculative-eagle-topk",
+                "1",
+                "--speculative-num-draft-tokens",
+                "4",
+                "--chunked-prefill-size",
+                "4096",
+                "--disable-flashinfer-autotune",
+            ],
+        )
+
+    @classmethod
+    def tearDownClass(cls):
+        if hasattr(cls, "process") and cls.process:
+            kill_process_tree(cls.process.pid)
+
+    def test_gsm8k(self):
+        _gsm8k_check(self)
+
+
 class TestDSV4FlashFP4B200Balanced(ServerSanityMixin, CustomTestCase):
    """Balanced recipe: TP=4, DP=4, DeepEP, EAGLE (1-step spec)."""

@@ -94,8 +138,8 @@ class TestDSV4FlashFP4B200Balanced(ServerSanityMixin, CustomTestCase):
        _gsm8k_check(self)


-class TestDSV4FlashFP4B200MaxThroughput(ServerSanityMixin, CustomTestCase):
-    """MaxThroughput recipe: TP=4, DP=4, DeepEP, no speculation."""
+class TestDSV4FlashFP4B200MegaMoE(ServerSanityMixin, CustomTestCase):
+    """Balanced recipe: TP=4, DP=4, MegaMoE."""

    @classmethod
    def setUpClass(cls):
@@ -114,10 +158,16 @@ class TestDSV4FlashFP4B200MaxThroughput(ServerSanityMixin, CustomTestCase):
                "--enable-dp-attention",
                "--moe-a2a-backend",
                "deepep",
-                "--deepep-config",
-                DEEPEP_CONFIG,
+                "--speculative-algorithm",
+                "EAGLE",
+                "--speculative-num-steps",
+                "1",
+                "--speculative-eagle-topk",
+                "1",
+                "--speculative-num-draft-tokens",
+                "2",
            ],
-            env=_DEEPEP_ENV,
+            env=_MEGAMOE_ENV,
        )

    @classmethod
@@ -24,7 +24,9 @@ from sglang.test.test_utils import (
 register_cuda_ci(est_time=900, suite="stage-c-test-dsv4-8-gpu-h200")

 MODEL = "deepseek-ai/DeepSeek-V4-Flash"
+MODEL_FP8 = "sgl-project/DeepSeek-V4-Flash-FP8"
 SERVER_LAUNCH_TIMEOUT = 3600
+DEEPEP_CONFIG = '{"normal_dispatch":{"num_sms":96},"normal_combine":{"num_sms":96}}'


 class TestDSV4FlashFP4H200(ServerSanityMixin, CustomTestCase):
@@ -1,10 +1,11 @@
-"""B200 per-commit CI: DeepSeek-V4-Flash FP4 (LowLatency recipe).
+"""H200 per-commit CI: DeepSeek-V4-Flash FP8 (LowLatency recipe).

-Launches TP=4 with flashinfer_mxfp4 MoE runner + EAGLE speculative decoding.
+Launches TP=4 with DeepEP a2a backend + EAGLE speculative decoding,
+with FP4 experts disabled via SGLANG_DSV4_FP4_EXPERTS=0.
 Runs 12 ServerSanity probes (correctness, streaming, concurrency, determinism)
 plus a GSM8K accuracy gate.

-Registry: stage-c-test-dsv4-4-gpu-b200 (per-commit, 4x B200)
+Registry: stage-c-test-dsv4-8-gpu-h200 (per-commit, 8x H200 — only 4 used by TP=4)
 """

 import unittest
@@ -21,18 +22,19 @@ from sglang.test.test_utils import (
    try_cached_model,
 )

-register_cuda_ci(est_time=900, suite="stage-c-test-dsv4-4-gpu-b200")
+register_cuda_ci(est_time=900, suite="stage-c-test-dsv4-8-gpu-h200")

-MODEL = "deepseek-ai/DeepSeek-V4-Flash"
+MODEL_FP8 = "sgl-project/DeepSeek-V4-Flash-FP8"
 SERVER_LAUNCH_TIMEOUT = 3600
+DEEPEP_CONFIG = '{"normal_dispatch":{"num_sms":96},"normal_combine":{"num_sms":96}}'


-class TestDSV4FlashFP4B200(ServerSanityMixin, CustomTestCase):
-    """LowLatency recipe: TP=4, FP4 (mxfp4), EAGLE spec decoding."""
+class TestDSV4FlashFP8H200(ServerSanityMixin, CustomTestCase):
+    """LowLatency recipe: TP=4, Marlin FP4, EAGLE spec decoding."""

    @classmethod
    def setUpClass(cls):
-        cls.model = try_cached_model(MODEL)
+        cls.model = try_cached_model(MODEL_FP8)
        cls.base_url = DEFAULT_URL_FOR_TEST
        cls.process = popen_launch_server(
            cls.model,
@@ -42,20 +44,30 @@ class TestDSV4FlashFP4B200(ServerSanityMixin, CustomTestCase):
                "--trust-remote-code",
                "--tp",
                "4",
-                "--moe-runner-backend",
-                "flashinfer_mxfp4",
+                "--dp",
+                "4",
+                "--enable-dp-attention",
+                "--moe-a2a-backend",
+                "deepep",
                "--speculative-algorithm",
                "EAGLE",
                "--speculative-num-steps",
-                "3",
+                "1",
                "--speculative-eagle-topk",
                "1",
                "--speculative-num-draft-tokens",
-                "4",
-                "--chunked-prefill-size",
-                "4096",
-                "--disable-flashinfer-autotune",
+                "2",
+                "--cuda-graph-max-bs",
+                "128",
+                "--max-running-requests",
+                "128",
+                "--deepep-config",
+                DEEPEP_CONFIG,
            ],
+            env={
+                "SGLANG_DSV4_FP4_EXPERTS": "0",
+                "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": "256",
+            },
        )

    @classmethod
@@ -74,7 +86,7 @@ class TestDSV4FlashFP4B200(ServerSanityMixin, CustomTestCase):
            num_threads=128,
        )
        metrics = run_eval(args)
-        print(f"[DSV4 Flash FP4 B200] GSM8K {metrics=}")
+        print(f"[DSV4 Flash FP4 Marlin H200] GSM8K {metrics=}")
        self.assertGreater(metrics["score"], 0.93)


@@ -220,6 +220,37 @@ class TestChatCompletionRequest(unittest.TestCase):
        self.assertFalse(request.chat_template_kwargs.get("thinking"))
        self.assertFalse(request.chat_template_kwargs.get("enable_thinking"))

+    def test_chat_completion_reasoning_effort_max(self):
+        """`max` is an sglang extension on chat completion's top-level
+        `reasoning_effort` only; the Responses-API-style nested
+        `reasoning.effort` path stays aligned with OpenAI's three levels."""
+        from pydantic import ValidationError
+
+        messages = [{"role": "user", "content": "Hello"}]
+        request = ChatCompletionRequest(
+            model="test-model",
+            messages=messages,
+            reasoning_effort="max",
+        )
+        self.assertEqual(request.reasoning_effort, "max")
+
+        # Unknown values still rejected.
+        with self.assertRaises(ValidationError):
+            ChatCompletionRequest(
+                model="test-model",
+                messages=messages,
+                reasoning_effort="ultra",
+            )
+
+        # Nested reasoning.effort=max is NOT promoted by normalize_reasoning_inputs:
+        # the Responses API path keeps the OpenAI low/medium/high contract.
+        request = ChatCompletionRequest(
+            model="test-model",
+            messages=messages,
+            reasoning={"effort": "max"},
+        )
+        self.assertNotEqual(request.reasoning_effort, "max")
+
    def test_chat_completion_json_format(self):
        """Test chat completion json format"""
        transcript = "Good morning! It's 7:00 AM, and I'm just waking up. Today is going to be a busy day, "
@@ -31,7 +31,7 @@ from sglang.srt.function_call.pythonic_detector import PythonicDetector
 from sglang.srt.function_call.qwen3_coder_detector import Qwen3CoderDetector
 from sglang.test.ci.ci_register import register_cpu_ci

-register_cpu_ci(15, "stage-a-test-cpu")
+register_cpu_ci(est_time=15, suite="stage-a-test-cpu")


 class TestPythonicDetector(unittest.TestCase):
@@ -1686,6 +1686,26 @@ class TestDeepSeekV32Detector(unittest.TestCase):
        grammar = xgr.Grammar.from_structural_tag(structural_tag)
        self.assertIsInstance(grammar, xgr.Grammar)

+    def test_self_closing_zero_arg_invoke(self):
+        """V32 inherits the same regex; verify self-closing parses to empty
+        params here too (V32 model rarely emits this shape, but the parser
+        must agree with V4 since V4 inherits from V32)."""
+        submit_tool = Tool(
+            type="function",
+            function=Function(
+                name="submit",
+                parameters={"type": "object", "properties": {}},
+            ),
+        )
+        text = (
+            '<｜DSML｜function_calls>\n<｜DSML｜invoke name="submit"/>\n'
+            "</｜DSML｜function_calls>"
+        )
+        result = self.detector.detect_and_parse(text, [submit_tool])
+        self.assertEqual(len(result.calls), 1)
+        self.assertEqual(result.calls[0].name, "submit")
+        self.assertEqual(json.loads(result.calls[0].parameters), {})
+

 class TestDeepSeekV4Detector(unittest.TestCase):
    def setUp(self):
@@ -2111,6 +2131,96 @@ class TestDeepSeekV4Detector(unittest.TestCase):
        grammar = xgr.Grammar.from_structural_tag(structural_tag)
        self.assertIsInstance(grammar, xgr.Grammar)

+    def test_self_closing_zero_arg_invoke(self):
+        """V4 emits `<｜DSML｜invoke name="x"/>` for zero-arg tools; the
+        detector must parse it as a complete tool call with empty params
+        instead of leaking the raw markup back into normal_text."""
+        submit_tool = Tool(
+            type="function",
+            function=Function(
+                name="submit",
+                description="Submit the final answer.",
+                parameters={"type": "object", "properties": {}},
+            ),
+        )
+
+        text = (
+            "Final answer.\n"
+            '<｜DSML｜tool_calls>\n<｜DSML｜invoke name="submit"/>\n'
+            "</｜DSML｜tool_calls>"
+        )
+        result = self.detector.detect_and_parse(text, [submit_tool])
+        self.assertEqual(len(result.calls), 1)
+        self.assertEqual(result.calls[0].name, "submit")
+        self.assertEqual(json.loads(result.calls[0].parameters), {})
+        self.assertNotIn("DSML", result.normal_text)
+
+    def test_self_closing_mixed_with_long_form(self):
+        """Mix of long-form (with params) and self-closing tags in one block."""
+        submit_tool = Tool(
+            type="function",
+            function=Function(
+                name="submit",
+                parameters={"type": "object", "properties": {}},
+            ),
+        )
+        text = (
+            "<｜DSML｜tool_calls>\n"
+            '<｜DSML｜invoke name="get_favorite_tourist_spot">\n'
+            '<｜DSML｜parameter name="city" string="true">SF</｜DSML｜parameter>\n'
+            "</｜DSML｜invoke>\n"
+            '<｜DSML｜invoke name="submit"/>\n'
+            "</｜DSML｜tool_calls>"
+        )
+        result = self.detector.detect_and_parse(text, self.tools + [submit_tool])
+        self.assertEqual(len(result.calls), 2)
+        self.assertEqual(result.calls[0].name, "get_favorite_tourist_spot")
+        self.assertEqual(json.loads(result.calls[0].parameters), {"city": "SF"})
+        self.assertEqual(result.calls[1].name, "submit")
+        self.assertEqual(json.loads(result.calls[1].parameters), {})
+
+    def test_streaming_self_closing_invoke(self):
+        """Self-closing invoke must terminate cleanly even when `/>` arrives
+        after the `name=` attribute crosses chunk boundaries."""
+        submit_tool = Tool(
+            type="function",
+            function=Function(
+                name="submit",
+                parameters={"type": "object", "properties": {}},
+            ),
+        )
+        # Build the prompt and feed it through the tokenizer to exercise the
+        # same chunk shapes the runtime sees.
+        text = (
+            "<｜DSML｜tool_calls>\n"
+            '<｜DSML｜invoke name="submit"/>\n'
+            "</｜DSML｜tool_calls>"
+        )
+        self.detector = DeepSeekV4Detector()
+        input_ids = self.tokenizer.encode(text, add_special_tokens=False)
+        chunks = [
+            self.tokenizer.decode(input_ids[i : i + self.interval])
+            for i in range(0, len(input_ids), self.interval)
+        ]
+
+        tool_calls_by_index = {}
+        for chunk in chunks:
+            result = self.detector.parse_streaming_increment(chunk, [submit_tool])
+            for call in result.calls:
+                if call.tool_index is None:
+                    continue
+                slot = tool_calls_by_index.setdefault(
+                    call.tool_index, {"name": "", "parameters": ""}
+                )
+                if call.name:
+                    slot["name"] = call.name
+                if call.parameters:
+                    slot["parameters"] += call.parameters
+
+        self.assertEqual(len(tool_calls_by_index), 1)
+        self.assertEqual(tool_calls_by_index[0]["name"], "submit")
+        self.assertEqual(json.loads(tool_calls_by_index[0]["parameters"]), {})
+

 class TestQwen3CoderDetector(unittest.TestCase):
    """Test suite for Qwen3CoderDetector."""