/test/registered/unit/test_model_overrides.py

"""Unit tests for the model-override machinery: whitelist metadata, registry,
gate, publish wiring, and the per-arch golden diffs for migrated families."""

from sglang.test.ci.ci_register import register_cpu_ci

register_cpu_ci(est_time=30, suite="base-a-test-cpu")

import dataclasses
import json
import os
import shutil
import tempfile
import unittest
from types import SimpleNamespace
from typing import Optional
from unittest.mock import patch

from sglang.srt.arg_groups import overrides as overrides_module
from sglang.srt.arg_groups.arg_utils import A, Arg, resolvable_fields
from sglang.srt.arg_groups.overrides import (
    collect_model_override_declarations,
    register_model_override,
    validate_declarations,
)
from sglang.srt.runtime_context import (
    get_context,
    get_server_args,
    reset_context,
)
from sglang.test.test_utils import CustomTestCase


@dataclasses.dataclass
class _FakeArgs:
    plain: A[int, "help text only"] = 0
    resolved_by_model: A[str, Arg(help="x", resolvable=True)] = "auto"
    also_resolved: A[Optional[int], Arg(help="y", resolvable=True)] = None
    metadata_but_not_overridable: A[bool, Arg(help="z")] = False


class TestModelOverridableWhitelist(CustomTestCase):
    def test_whitelist_derivation_from_annotated_metadata(self):
        self.assertEqual(
            resolvable_fields(_FakeArgs),
            frozenset({"resolved_by_model", "also_resolved"}),
        )

    def test_server_args_whitelist_is_exactly_the_migrated_fields(self):
        # Fields are whitelisted one family at a time by the migration
        # sweeps. This pin makes accidental tagging visible — extend it in
        # the same commit that tags a new field.
        from sglang.srt.server_args import ServerArgs

        self.assertEqual(
            resolvable_fields(ServerArgs),
            frozenset(
                {
                    "dtype",
                    "enable_tf32_matmul",
                    "enable_multi_layer_eagle",
                    "swa_full_tokens_ratio",
                    "disable_hybrid_swa_memory",
                    "sampling_backend",
                    "attention_backend",
                    "page_size",
                    "moe_runner_backend",
                    "quantization",
                    "enable_dp_attention",
                    "enable_dp_lm_head",
                    "moe_a2a_backend",
                    "ep_size",
                    "moe_dense_tp_size",
                    "attn_cp_size",
                    "disable_overlap_schedule",
                    "uses_mamba_radix_cache",
                    "mamba_radix_cache_strategy",
                    "speculative_moe_runner_backend",
                    "speculative_moe_a2a_backend",
                    "disable_shared_experts_fusion",
                    "kv_cache_dtype",
                    "dsa_prefill_backend",
                    "dsa_decode_backend",
                    "prefill_attention_backend",
                    "decode_attention_backend",
                    "flashinfer_allreduce_fusion_backend",
                    "fp8_gemm_runner_backend",
                }
            ),
        )


class _IsolatedRegistry(CustomTestCase):
    """Run each test against empty registries (they are process-global)."""

    def setUp(self):
        super().setUp()
        self._patches = [
            patch.dict(overrides_module.MODEL_OVERRIDES, clear=True),
            patch.dict(overrides_module._MODEL_OVERRIDE_FNS, clear=True),
            patch.object(overrides_module, "_PREDICATE_OVERRIDE_FNS", []),
        ]
        for p in self._patches:
            p.start()

    def tearDown(self):
        for p in self._patches:
            p.stop()
        super().tearDown()


class TestModelOverrideRegistry(_IsolatedRegistry):
    def test_const_then_callables_in_registration_order(self):
        overrides_module.MODEL_OVERRIDES["FakeForCausalLM"] = {"a": 1}

        @register_model_override("FakeForCausalLM")
        def _first(server_args, hf_config):
            return {"b": server_args.base + 1}

        @register_model_override("FakeForCausalLM")
        def _second(server_args, hf_config):
            return {"a": 3}

        declarations = collect_model_override_declarations(
            "FakeForCausalLM", SimpleNamespace(base=10), hf_config=None
        )
        self.assertEqual(
            declarations,
            [
                ("MODEL_OVERRIDES['FakeForCausalLM']", {"a": 1}),
                (_first.__qualname__, {"b": 11}),
                (_second.__qualname__, {"a": 3}),
            ],
        )

    def test_unknown_architecture_yields_nothing(self):
        self.assertEqual(
            collect_model_override_declarations("NoSuchArch", None, None), []
        )

    def test_empty_declarations_are_dropped(self):
        @register_model_override("FakeForCausalLM")
        def _nothing_applies(server_args, hf_config):
            return {}

        self.assertEqual(
            collect_model_override_declarations("FakeForCausalLM", None, None), []
        )

    def test_non_dict_return_is_rejected(self):
        @register_model_override("FakeForCausalLM")
        def _bad(server_args, hf_config):
            return None

        with self.assertRaises(TypeError):
            collect_model_override_declarations("FakeForCausalLM", None, None)

    def test_predicate_keyed_provider(self):
        from sglang.srt.arg_groups.overrides import register_model_override_predicate

        @register_model_override("FakeStep9ForCausalLM")
        def _exact(server_args, hf_config):
            return {"a": 1}

        @register_model_override_predicate(lambda arch: "Step9" in arch)
        def _by_predicate(server_args, hf_config):
            return {"b": 2}

        # matching arch: exact-keyed first, then predicate-keyed
        self.assertEqual(
            collect_model_override_declarations("FakeStep9ForCausalLM", None, None),
            [(_exact.__qualname__, {"a": 1}), (_by_predicate.__qualname__, {"b": 2})],
        )
        # non-matching arch: predicate does not fire
        self.assertEqual(
            collect_model_override_declarations("OtherForCausalLM", None, None), []
        )


class TestResolvedViewAndPasses(CustomTestCase):
    """Pipeline skeleton: read-only view semantics + transition invocation."""

    def test_view_forwards_reads_and_rejects_writes(self):
        from sglang.srt.arg_groups.overrides import ResolvedView

        live = SimpleNamespace(a=1, method=lambda: "m")
        view = ResolvedView(live)
        self.assertEqual(view.a, 1)
        self.assertEqual(view.method(), "m")  # method forwarding
        live.a = 2
        self.assertEqual(view.a, 2)  # live, not a snapshot
        with self.assertRaises(AttributeError):
            view.a = 3

    def test_view_overlay_wins(self):
        from sglang.srt.arg_groups.overrides import ResolvedView

        view = ResolvedView(SimpleNamespace(a=1, b=2), overlay={"a": 10})
        self.assertEqual(view.a, 10)
        self.assertEqual(view.b, 2)

    def test_run_pass_appends_stash_and_stays_pristine(self):
        from sglang.srt.arg_groups.overrides import run_post_process_pass

        live = SimpleNamespace(x=None, _resolved_overrides=[])

        def _fill_x(view):
            return {"x": "filled"} if view.x is None else {}

        run_post_process_pass(live, _fill_x)
        self.assertIsNone(live.x)  # never applied in place
        self.assertEqual(
            live._resolved_overrides, [(_fill_x.__qualname__, {"x": "filled"})]
        )
        # the next invocation sees the declared value through the overlay
        run_post_process_pass(live, _fill_x)
        self.assertEqual(len(live._resolved_overrides), 1)

    def test_run_pass_rejects_non_dict(self):
        from sglang.srt.arg_groups.overrides import run_post_process_pass

        with self.assertRaises(TypeError):
            run_post_process_pass(
                SimpleNamespace(_resolved_overrides=[]), lambda view: None
            )


class _IsolatedPublish(CustomTestCase):
    """Publishing writes the process-global context; save/restore around it."""

    def setUp(self):
        super().setUp()
        self._saved_server_args = get_context()._server_args

    def tearDown(self):
        reset_context()
        if self._saved_server_args is not None:
            get_context()._server_args = self._saved_server_args
        super().tearDown()


class TestPublishInstallsSlot(_IsolatedPublish):
    """Publish wiring: set_server_args installs the already-resolved object
    into the context-owned slot (no transformation at publish time)."""

    def test_dummy_fixture_has_empty_stash_and_publishes_cleanly(self):
        from sglang.srt.server_args import (
            ServerArgs,
            set_global_server_args_for_scheduler,
        )

        sa = ServerArgs(model_path="dummy")  # __post_init__ early-returns
        # The stash is created before the dummy short-circuit and stays empty.
        self.assertEqual(sa._resolved_overrides, [])
        set_global_server_args_for_scheduler(sa)
        self.assertIs(get_server_args(), sa)


class TestGoldenModelOverrides(_IsolatedPublish):
    """Per-arch golden diff for migrated families: the declarative path must
    reproduce the legacy imperative writes byte-identically on the
    materialized server_args fields; the publish round-trip returns the same
    object."""

    _MINI_CONFIG = {
        "hidden_size": 64,
        "intermediate_size": 128,
        "num_attention_heads": 4,
        "num_hidden_layers": 2,
        "num_key_value_heads": 2,
        "vocab_size": 512,
        "max_position_embeddings": 128,
        "rms_norm_eps": 1e-5,
        "torch_dtype": "bfloat16",
        # MLA shape fields (required by the MistralLarge3/Pixtral arch
        # family; inert extras for non-MLA control arches).
        "kv_lora_rank": 32,
        "qk_nope_head_dim": 16,
        "qk_rope_head_dim": 8,
        "v_head_dim": 16,
    }

    def _construct(self, arch, model_type, config_extra=None, **server_kwargs):
        from sglang.srt.server_args import ServerArgs

        # Golden resolution must be host-independent: accelerator-less CI
        # runners resolve only the base platform, where get_device() raises.
        server_kwargs.setdefault("device", "cuda")
        config = dict(self._MINI_CONFIG, architectures=[arch], model_type=model_type)
        config.update(config_extra or {})
        config_dir = tempfile.mkdtemp(prefix="golden_override_")
        self.addCleanup(shutil.rmtree, config_dir, ignore_errors=True)
        with open(os.path.join(config_dir, "config.json"), "w") as f:
            json.dump(config, f)
        return ServerArgs(model_path=config_dir, **server_kwargs)

    def _publish(self, server_args):
        from sglang.srt.server_args import (
            set_global_server_args_for_scheduler,
        )

        set_global_server_args_for_scheduler(server_args)
        return get_server_args()

    def test_mistral_large3_forces_bfloat16(self):
        sa = self._construct("MistralLarge3ForCausalLM", "mistral")
        self.assertEqual(sa.dtype, "bfloat16")  # materialized at end of resolution
        self.assertIn(
            ("MODEL_OVERRIDES['MistralLarge3ForCausalLM']", {"dtype": "bfloat16"}),
            sa._resolved_overrides,
        )
        self.assertEqual(self._publish(sa).dtype, "bfloat16")

    def test_user_requested_dtype_is_still_overridden(self):
        # Legacy fidelity: the arch branch overwrote dtype unconditionally,
        # so the declaration must too. The pristine request survives on
        # provenance; the materialized field carries the override.
        sa = self._construct("MistralLarge3ForCausalLM", "mistral", dtype="float16")
        self.assertEqual(sa.dtype, "bfloat16")  # materialized
        self.assertEqual(self._publish(sa).dtype, "bfloat16")

    def test_control_arch_keeps_pristine_dtype(self):
        sa = self._construct("LlamaForCausalLM", "llama")
        self.assertEqual(sa.dtype, "auto")
        declared = {f for _s, d in sa._resolved_overrides for f in d}
        self.assertNotIn("dtype", declared)  # no arch declaration for Llama
        # publish still materializes the whitelisted leaf with the pristine
        # value: readers only ever read flags.
        self.assertEqual(self._publish(sa).dtype, "auto")

    def test_minimax_m2_enables_tf32_matmul(self):
        sa = self._construct("MiniMaxM2ForCausalLM", "llama")
        self.assertTrue(sa.enable_tf32_matmul)  # materialized
        self.assertIn(
            ("_minimax_m2_overrides", {"enable_tf32_matmul": True}),
            sa._resolved_overrides,
        )
        flags = self._publish(sa)
        self.assertTrue(flags.enable_tf32_matmul)
        self.assertFalse(flags.enable_multi_layer_eagle)  # pristine materialize

    def test_mimo_v2_declarations(self):
        # Callable-level golden: MiMoV2 archs are hybrid (config-shape heavy),
        # so the declaration is pinned directly for both provider inputs.
        from sglang.srt.arg_groups.overrides import _mimo_v2_overrides

        self.assertEqual(
            _mimo_v2_overrides(SimpleNamespace(speculative_algorithm="EAGLE"), None),
            {"enable_multi_layer_eagle": True},
        )
        self.assertEqual(
            _mimo_v2_overrides(SimpleNamespace(speculative_algorithm=None), None),
            {},
        )

    def test_mimo_v2_family_is_registered(self):
        self.assertEqual(
            collect_model_override_declarations(
                "MiMoV2FlashForCausalLM",
                SimpleNamespace(speculative_algorithm="EAGLE"),
                None,
            ),
            [("_mimo_v2_overrides", {"enable_multi_layer_eagle": True})],
        )

    def test_step3p_hierarchical_cache_golden(self):
        # SWA-hybrid arch: the mini config needs layer_types/sliding_window.
        config_extra = {
            "layer_types": ["sliding_attention", "full_attention"],
            "sliding_window": 64,
        }
        sa = self._construct(
            "Step3p5ForCausalLM",
            "llama",
            config_extra=config_extra,
            enable_hierarchical_cache=True,
        )
        # materialized at the end of resolution
        self.assertEqual(sa.swa_full_tokens_ratio, 1.0)
        self.assertTrue(sa.disable_hybrid_swa_memory)
        flags = self._publish(sa)
        self.assertEqual(flags.swa_full_tokens_ratio, 1.0)
        self.assertTrue(flags.disable_hybrid_swa_memory)

    def test_gemma2_disables_hybrid_swa_memory(self):
        sa = self._construct("Gemma2ForCausalLM", "llama")
        self.assertTrue(sa.disable_hybrid_swa_memory)  # materialized
        self.assertIn(
            ("_gemma2_gemma3_overrides", {"disable_hybrid_swa_memory": True}),
            sa._resolved_overrides,
        )
        self.assertTrue(self._publish(sa).disable_hybrid_swa_memory)

    def test_olmo2_disables_hybrid_swa_memory(self):
        sa = self._construct("Olmo2ForCausalLM", "llama")
        self.assertTrue(sa.disable_hybrid_swa_memory)  # materialized
        self.assertTrue(self._publish(sa).disable_hybrid_swa_memory)

    def test_exaone_conditional_on_sliding_window_pattern(self):
        # With the pattern the branch also asserts an explicit backend.
        sa = self._construct(
            "Exaone4ForCausalLM",
            "llama",
            config_extra={"sliding_window_pattern": "LLLG"},
            attention_backend="fa3",
        )
        self.assertTrue(sa.disable_hybrid_swa_memory)  # materialized
        self.assertTrue(self._publish(sa).disable_hybrid_swa_memory)

    def test_exaone_without_pattern_declares_nothing(self):
        from sglang.srt.arg_groups.overrides import _exaone_overrides

        self.assertEqual(
            _exaone_overrides(None, SimpleNamespace(sliding_window_pattern=None)),
            {},
        )

    def test_gpt_oss_mxfp4_forces_bfloat16(self):
        from sglang.srt.layers.quantization import QUANTIZATION_METHODS

        if "mxfp4" not in QUANTIZATION_METHODS:
            # Registration is platform-gated (CUDA / CPU engine / MXFP-HIP);
            # plain CPU CI runners cannot construct an mxfp4 ModelConfig.
            self.skipTest("mxfp4 quantization is not registered on this platform")
        sa = self._construct(
            "GptOssForCausalLM",
            "llama",
            config_extra={"quantization_config": {"quant_method": "mxfp4"}},
        )
        self.assertEqual(sa.dtype, "bfloat16")  # materialized
        self.assertEqual(self._publish(sa).dtype, "bfloat16")

    def test_gpt_oss_without_mxfp4_keeps_pristine_dtype(self):
        sa = self._construct("GptOssForCausalLM", "llama")
        self.assertEqual(sa.dtype, "auto")
        self.assertEqual(self._publish(sa).dtype, "auto")

    def test_gpt_oss_xpu_dtype_validation_reads_pristine(self):
        from sglang.srt.arg_groups.overrides import _gpt_oss_overrides

        with patch.object(overrides_module, "is_xpu", return_value=True):
            with self.assertRaises(NotImplementedError):
                _gpt_oss_overrides(
                    SimpleNamespace(
                        dtype="float16",
                        is_attention_backend_not_set=lambda: False,
                    ),
                    SimpleNamespace(architectures=["GptOssForCausalLM"]),
                )

    def test_sampling_backend_default_pass(self):
        from sglang.srt.utils.common import is_flashinfer_available

        sa = self._construct("LlamaForCausalLM", "llama")
        expected = "flashinfer" if is_flashinfer_available() else "pytorch"
        self.assertEqual(sa.sampling_backend, expected)  # materialized
        self.assertIn(
            ("_sampling_backend_default", {"sampling_backend": expected}),
            sa._resolved_overrides,
        )
        self.assertEqual(self._publish(sa).sampling_backend, expected)

    def test_sampling_backend_user_choice_survives(self):
        sa = self._construct("LlamaForCausalLM", "llama", sampling_backend="pytorch")
        self.assertEqual(sa.sampling_backend, "pytorch")
        # the pass declared nothing; publish materializes the pristine choice
        self.assertEqual(self._publish(sa).sampling_backend, "pytorch")

    def test_deterministic_inference_forces_pytorch_sampling(self):
        sa = self._construct(
            "LlamaForCausalLM", "llama", enable_deterministic_inference=True
        )
        # two pass writers chain: default fill, then the deterministic force —
        # last writer wins; materialization lands the end state on the fields.
        self.assertEqual(sa.sampling_backend, "pytorch")
        flags = self._publish(sa)
        self.assertEqual(flags.sampling_backend, "pytorch")
        # the deterministic attention fill declared a compatible backend and
        # the compatibility default-fill then had nothing to do
        deterministic_fills = [
            decl["attention_backend"]
            for source, decl in sa._resolved_overrides
            if source == "_deterministic_attention_backend"
        ]
        self.assertEqual(len(deterministic_fills), 1)
        self.assertEqual(sa.attention_backend, deterministic_fills[0])
        self.assertEqual(flags.attention_backend, deterministic_fills[0])

    def test_deterministic_incompatible_backend_raises(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deterministic_attention_backend,
        )

        view = ResolvedView(
            SimpleNamespace(
                enable_deterministic_inference=True, attention_backend="flashmla"
            )
        )
        with self.assertRaises(ValueError):
            _deterministic_attention_backend(view)

    def test_deterministic_ascend_is_left_alone(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deterministic_sampling_backend,
        )

        view = ResolvedView(
            SimpleNamespace(
                enable_deterministic_inference=True, sampling_backend="ascend"
            )
        )
        self.assertEqual(_deterministic_sampling_backend(view), {})

    def test_dllm_forces_flashinfer_with_cuda_graph(self):
        # CUDA path: cuda graph enabled by default -> dllm forces flashinfer.
        # A real dllm arch: the page pass now runs regardless of the radix
        # switch and builds DllmConfig for it.
        sa = self._construct(
            "SDARForCausalLM",
            "llama",
            dllm_algorithm="LowConfidence",
            disable_radix_cache=True,
        )
        self.assertEqual(sa.attention_backend, "flashinfer")  # materialized
        self.assertIn(
            ("_dllm_attention_backend", {"attention_backend": "flashinfer"}),
            sa._resolved_overrides,
        )
        # the deterministic fill lands on the attention_backend field
        self.assertEqual(self._publish(sa).attention_backend, "flashinfer")

    def test_attention_backend_leaf_materializes_end_state(self):
        # The default-fill pass declares the platform-selected backend; the
        # leaf must equal the last declared value while the server_args field
        # stays pristine (dual-apply retired).
        sa = self._construct("LlamaForCausalLM", "llama")
        declared_values = [
            d["attention_backend"]
            for _s, d in sa._resolved_overrides
            if "attention_backend" in d
        ]
        self.assertTrue(declared_values)  # default fill declared
        self.assertEqual(sa.attention_backend, declared_values[-1])  # materialized
        self.assertEqual(self._publish(sa).attention_backend, declared_values[-1])

    def test_post_materialize_pass_writes_through(self):
        from sglang.srt.arg_groups.overrides import run_post_process_pass

        # A pass invoked after materialization (a post-init slot, like the
        # legacy runner-side adjustments) declares AND writes through, so
        # field readers and the publish see the same end state.
        sa = self._construct("LlamaForCausalLM", "llama")
        resolved_before = sa.attention_backend

        def _force_triton(view):
            if view.attention_backend != "triton":
                return {"attention_backend": "triton"}
            return {}

        run_post_process_pass(sa, _force_triton)
        if resolved_before != "triton":
            self.assertEqual(sa.attention_backend, "triton")
        self.assertEqual(self._publish(sa).attention_backend, sa.attention_backend)

    def test_attention_backend_user_choice_declares_nothing_extra(self):
        sa = self._construct("LlamaForCausalLM", "llama", attention_backend="triton")
        self.assertEqual(sa.attention_backend, "triton")
        self.assertEqual(self._publish(sa).attention_backend, "triton")

    def test_compatibility_passes_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _attention_backend_default,
            _attention_backend_dual_chunk,
            _attention_backend_fa3_fp8_fallback,
            _attention_backend_platform_fallbacks,
        )

        # split-backend override wins over the default fill
        view = ResolvedView(
            SimpleNamespace(
                prefill_attention_backend="fa3",
                decode_attention_backend="fa3",
                attention_backend=None,
            )
        )
        self.assertEqual(_attention_backend_default(view), {"attention_backend": "fa3"})

        # fa3 + fp8_e5m2 falls back to triton
        view = ResolvedView(
            SimpleNamespace(attention_backend="fa3", kv_cache_dtype="fp8_e5m2")
        )
        self.assertEqual(
            _attention_backend_fa3_fp8_fallback(view),
            {"attention_backend": "triton"},
        )

        # amx fallback fires only without hardware support
        view = ResolvedView(
            SimpleNamespace(attention_backend="intel_amx", device="cpu")
        )
        with patch.object(overrides_module, "cpu_has_amx_support", return_value=False):
            self.assertEqual(
                _attention_backend_platform_fallbacks(view),
                {"attention_backend": "torch_native"},
            )
        with patch.object(overrides_module, "cpu_has_amx_support", return_value=True):
            self.assertEqual(_attention_backend_platform_fallbacks(view), {})

        # dual-chunk config: mismatched explicit backend raises verbatim
        def _mc(dual):
            return SimpleNamespace(
                get_model_config=lambda: SimpleNamespace(
                    hf_config=SimpleNamespace(dual_chunk_attention_config=dual)
                ),
                attention_backend="fa3",
            )

        with self.assertRaises(ValueError):
            _attention_backend_dual_chunk(ResolvedView(_mc({"a": 1})))
        self.assertEqual(_attention_backend_dual_chunk(ResolvedView(_mc(None))), {})

    def test_dllm_platform_paths_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _dllm_attention_backend,
        )
        from sglang.srt.model_executor.cuda_graph_config import Backend

        def _view(**kw):
            defaults = dict(
                dllm_algorithm="LowConfidence",
                attention_backend=None,
                cuda_graph_config=SimpleNamespace(
                    decode=SimpleNamespace(backend=Backend.DISABLED)
                ),
            )
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        with patch.object(overrides_module, "is_hip", return_value=True):
            self.assertEqual(
                _dllm_attention_backend(_view()), {"attention_backend": "triton"}
            )
            self.assertEqual(
                _dllm_attention_backend(_view(attention_backend="aiter")), {}
            )
        with patch.object(overrides_module, "is_hip", return_value=False):
            with patch.object(overrides_module, "is_npu", return_value=True):
                self.assertEqual(
                    _dllm_attention_backend(_view()),
                    {"attention_backend": "ascend"},
                )
            with patch.object(overrides_module, "is_npu", return_value=False):
                # cuda graph disabled -> nothing to force
                self.assertEqual(_dllm_attention_backend(_view()), {})
                self.assertEqual(
                    _dllm_attention_backend(_view(dllm_algorithm=None)), {}
                )

    def test_page_size_default_pass(self):
        from sglang.srt.arg_groups.overrides import ResolvedView, _page_size_default

        # user-set page_size: nothing to declare
        self.assertEqual(
            _page_size_default(ResolvedView(SimpleNamespace(page_size=64))), {}
        )
        # default fill on non-HIP/non-MUSA platforms is 1
        with patch.object(overrides_module, "is_hip", return_value=False):
            with patch.object(overrides_module, "is_musa", return_value=False):
                self.assertEqual(
                    _page_size_default(ResolvedView(SimpleNamespace(page_size=None))),
                    {"page_size": 1},
                )
            with patch.object(overrides_module, "is_musa", return_value=True):
                self.assertEqual(
                    _page_size_default(ResolvedView(SimpleNamespace(page_size=None))),
                    {"page_size": 64},
                )

    def test_dllm_page_size_pass(self):
        from sglang.srt.arg_groups.overrides import ResolvedView, _dllm_page_size

        def _view(**kw):
            defaults = dict(
                dllm_algorithm="LowConfidence", disable_radix_cache=False, page_size=1
            )
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        with patch(
            "sglang.srt.dllm.config.DllmConfig.from_server_args",
            return_value=SimpleNamespace(block_size=32),
        ):
            self.assertEqual(_view() and _dllm_page_size(_view()), {"page_size": 32})
            # aligned but larger than the block: the scheduler-init fallback
            # (folded into this pass) still caps the page at the block size
            self.assertEqual(_dllm_page_size(_view(page_size=64)), {"page_size": 32})
            self.assertEqual(_dllm_page_size(_view(page_size=32)), {})  # equal
            # radix disabled skips the alignment fill but keeps the cap
            self.assertEqual(_dllm_page_size(_view(disable_radix_cache=True)), {})
            self.assertEqual(
                _dllm_page_size(_view(disable_radix_cache=True, page_size=64)),
                {"page_size": 32},
            )
        self.assertEqual(_dllm_page_size(_view(dllm_algorithm=None)), {})

    def test_overlap_disable_passes(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _dllm_overlap_disable,
            _pipeline_parallel_overlap_disable,
            _sparse_head_overlap_disable,
        )

        # pipeline parallelism: declares only when pp_size > 1
        self.assertEqual(
            _pipeline_parallel_overlap_disable(
                ResolvedView(SimpleNamespace(pp_size=1))
            ),
            {},
        )
        self.assertEqual(
            _pipeline_parallel_overlap_disable(
                ResolvedView(SimpleNamespace(pp_size=2))
            ),
            {"disable_overlap_schedule": True},
        )

        # dllm: guarded on the algorithm and the current value
        def _view(**kw):
            defaults = dict(
                dllm_algorithm="LowConfidence", disable_overlap_schedule=False
            )
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        self.assertEqual(_dllm_overlap_disable(_view(dllm_algorithm=None)), {})
        self.assertEqual(
            _dllm_overlap_disable(_view(disable_overlap_schedule=True)), {}
        )
        self.assertEqual(
            _dllm_overlap_disable(_view()), {"disable_overlap_schedule": True}
        )

        # embeddings sparse head: keyed on the env var being set
        from sglang.srt.environ import envs

        view = ResolvedView(SimpleNamespace())
        with patch.object(
            envs.SGLANG_EMBEDDINGS_SPARSE_HEAD, "is_set", return_value=False
        ):
            self.assertEqual(_sparse_head_overlap_disable(view), {})
        with patch.object(
            envs.SGLANG_EMBEDDINGS_SPARSE_HEAD, "is_set", return_value=True
        ):
            self.assertEqual(
                _sparse_head_overlap_disable(view), {"disable_overlap_schedule": True}
            )

    def test_deepseek_v4_overrides_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import _deepseek_v4_overrides
        from sglang.srt.server_args import ServerArgs

        hf = SimpleNamespace(architectures=["DeepseekV4ForCausalLM"])

        def _args(**kw):
            defaults = dict(
                device="cuda",
                swa_full_tokens_ratio=ServerArgs.swa_full_tokens_ratio,
                moe_runner_backend="auto",
                get_model_config=lambda: SimpleNamespace(nvfp4_moe_meta=None),
            )
            defaults.update(kw)
            return SimpleNamespace(**defaults)

        self.assertEqual(
            _deepseek_v4_overrides(_args(), hf),
            {
                "attention_backend": "dsv4",
                "page_size": 256,
                "swa_full_tokens_ratio": 0.1,
            },
        )
        # NPU pool geometry
        self.assertEqual(
            _deepseek_v4_overrides(_args(device="npu"), hf)["page_size"], 128
        )
        # user-set window ratio survives
        self.assertNotIn(
            "swa_full_tokens_ratio",
            _deepseek_v4_overrides(_args(swa_full_tokens_ratio=0.5), hf),
        )
        # nvfp4 hybrid checkpoint routes the MoE runner
        self.assertEqual(
            _deepseek_v4_overrides(
                _args(
                    get_model_config=lambda: SimpleNamespace(nvfp4_moe_meta=object())
                ),
                hf,
            )["moe_runner_backend"],
            "flashinfer_trtllm_routed",
        )

    def test_deepseek_v4_sm120_moe_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deepseek_v4_sm120_moe,
        )

        def _view(arch="DeepseekV4ForCausalLM", **kw):
            hf = SimpleNamespace(architectures=[arch])
            defaults = dict(moe_runner_backend="auto")
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        with patch.object(overrides_module, "is_sm120_supported", return_value=True):
            self.assertEqual(
                _deepseek_v4_sm120_moe(_view()), {"moe_runner_backend": "marlin"}
            )
            self.assertEqual(
                _deepseek_v4_sm120_moe(_view(moe_runner_backend="triton")), {}
            )
            self.assertEqual(_deepseek_v4_sm120_moe(_view(arch="LlamaForCausalLM")), {})
        with patch.object(overrides_module, "is_sm120_supported", return_value=False):
            self.assertEqual(_deepseek_v4_sm120_moe(_view()), {})

    def test_nemotron_h_overrides_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import _nemotron_h_overrides

        def _hf(quant_algo="NVFP4"):
            return SimpleNamespace(
                architectures=["NemotronHForCausalLM"],
                mlp_hidden_act="relu2",
                quantization_config={"quant_algo": quant_algo},
            )

        def _args(mc_quant, hf, **kw):
            mc = SimpleNamespace(quantization=mc_quant, hf_config=hf)
            defaults = dict(
                quantization=None,
                moe_runner_backend="auto",
                moe_a2a_backend="none",
                attention_backend=None,
                get_model_config=lambda: mc,
            )
            defaults.update(kw)
            return SimpleNamespace(**defaults)

        hf = _hf()
        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            # modelopt checkpoint: quant algo resolution + sm100 defaults
            self.assertEqual(
                _nemotron_h_overrides(_args("modelopt", hf), hf),
                {
                    "quantization": "modelopt_fp4",
                    "moe_runner_backend": "flashinfer_trtllm",
                    "attention_backend": "flashinfer",
                },
            )
            hf_mixed = _hf("MIXED_PRECISION")
            self.assertEqual(
                _nemotron_h_overrides(_args("modelopt", hf_mixed), hf_mixed)[
                    "quantization"
                ],
                "modelopt_mixed",
            )
        with (
            patch.object(overrides_module, "is_sm100_supported", return_value=False),
            patch.object(overrides_module, "is_cuda", return_value=True),
            patch.object(
                overrides_module, "get_device_capability", return_value=(9, 0)
            ),
        ):
            # SM80-SM90 fp4: marlin
            self.assertEqual(
                _nemotron_h_overrides(_args("modelopt_fp4", hf), hf),
                {"quantization": "modelopt_fp4", "moe_runner_backend": "marlin"},
            )
            # unquantized checkpoint: cutlass fallback, no quant declared
            self.assertEqual(
                _nemotron_h_overrides(_args(None, hf), hf),
                {"moe_runner_backend": "flashinfer_cutlass"},
            )
            # non-modelopt quantized checkpoint: nothing declared
            self.assertEqual(_nemotron_h_overrides(_args("fp8", hf), hf), {})
            # user-set moe backend survives
            self.assertEqual(
                _nemotron_h_overrides(_args(None, hf, moe_runner_backend="triton"), hf),
                {},
            )

    def test_speculative_moe_runner_default_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _speculative_moe_runner_default,
        )

        self.assertEqual(
            _speculative_moe_runner_default(
                ResolvedView(
                    SimpleNamespace(
                        speculative_moe_runner_backend=None, moe_runner_backend="triton"
                    )
                )
            ),
            {"speculative_moe_runner_backend": "triton"},
        )
        # user-set draft backend survives
        self.assertEqual(
            _speculative_moe_runner_default(
                ResolvedView(
                    SimpleNamespace(
                        speculative_moe_runner_backend="deep_gemm",
                        moe_runner_backend="auto",
                    )
                )
            ),
            {},
        )

    def test_dsa_split_backend_resolution_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _dsa_split_backend_resolution,
        )

        def _view(arch="DeepseekV32ForCausalLM", **kw):
            hf = SimpleNamespace(architectures=[arch])
            defaults = dict(
                kv_cache_dtype="fp8_e4m3",
                dsa_prefill_backend=None,
                dsa_decode_backend=None,
                enable_hisparse=False,
            )
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        with (
            patch("sglang.srt.configs.model_config.is_deepseek_dsa", return_value=True),
            patch.object(overrides_module, "is_npu", return_value=False),
            patch.object(overrides_module, "is_xpu", return_value=False),
            patch.object(overrides_module, "is_hip", return_value=False),
            patch("torch.cuda.get_device_capability", return_value=(9, 0)),
        ):
            # Hopper FP8 -> flashmla_kv both
            self.assertEqual(
                _dsa_split_backend_resolution(_view()),
                {
                    "dsa_prefill_backend": "flashmla_kv",
                    "dsa_decode_backend": "flashmla_kv",
                },
            )
            # Hopper bf16 -> flashmla_sparse / fa3
            self.assertEqual(
                _dsa_split_backend_resolution(_view(kv_cache_dtype="bfloat16")),
                {
                    "dsa_prefill_backend": "flashmla_sparse",
                    "dsa_decode_backend": "fa3",
                },
            )
            # user-set prefill survives; only decode defaulted
            self.assertEqual(
                _dsa_split_backend_resolution(_view(dsa_prefill_backend="trtllm")),
                {"dsa_decode_backend": "flashmla_kv"},
            )
            # hisparse arm takes precedence (CUDA fp8 -> flashmla_kv)
            self.assertEqual(
                _dsa_split_backend_resolution(_view(enable_hisparse=True)),
                {
                    "dsa_prefill_backend": "flashmla_kv",
                    "dsa_decode_backend": "flashmla_kv",
                },
            )
            # non-family arch declares nothing
            self.assertEqual(
                _dsa_split_backend_resolution(_view(arch="LlamaForCausalLM")), {}
            )
        with (
            patch("sglang.srt.configs.model_config.is_deepseek_dsa", return_value=True),
            patch.object(overrides_module, "is_npu", return_value=False),
            patch.object(overrides_module, "is_xpu", return_value=False),
            patch.object(overrides_module, "is_hip", return_value=True),
            patch("torch.cuda.get_device_capability", return_value=(9, 4)),
        ):
            # ROCm with both unset -> tilelang
            self.assertEqual(
                _dsa_split_backend_resolution(_view(kv_cache_dtype="bfloat16")),
                {
                    "dsa_prefill_backend": "tilelang",
                    "dsa_decode_backend": "tilelang",
                },
            )

    def test_flashinfer_allreduce_fusion_passes(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deterministic_allreduce_fusion_disable,
            _enforce_disable_allreduce_fusion,
            _flashinfer_allreduce_fusion_auto_enable,
        )

        def _view(arch="Qwen3MoeForCausalLM", **kw):
            hf = SimpleNamespace(architectures=[arch])
            defaults = dict(
                flashinfer_allreduce_fusion_backend=None,
                tp_size=2,
                enable_dp_attention=False,
                nnodes=1,
                moe_a2a_backend="none",
                enforce_disable_flashinfer_allreduce_fusion=False,
                enable_deterministic_inference=False,
            )
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        with (
            patch.object(overrides_module, "is_sm90_supported", return_value=True),
            patch.object(overrides_module, "is_sm100_supported", return_value=False),
        ):
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(_view()),
                {"flashinfer_allreduce_fusion_backend": "auto"},
            )
            # guards: unsupported arch / tp==1 / dp attention / a2a backend
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(
                    _view(arch="LlamaForCausalLM")
                ),
                {},
            )
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(_view(tp_size=1)), {}
            )
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(
                    _view(enable_dp_attention=True)
                ),
                {},
            )
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(
                    _view(moe_a2a_backend="deepep")
                ),
                {},
            )
            # SM90 multi-node: blocked (nnodes>1 needs SM100)
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(_view(nnodes=2)), {}
            )
            # user-set backend survives
            self.assertEqual(
                _flashinfer_allreduce_fusion_auto_enable(
                    _view(flashinfer_allreduce_fusion_backend="trtllm")
                ),
                {},
            )

        # enforce-disable wins over everything
        self.assertEqual(
            _enforce_disable_allreduce_fusion(
                _view(
                    flashinfer_allreduce_fusion_backend="auto",
                    enforce_disable_flashinfer_allreduce_fusion=True,
                )
            ),
            {"flashinfer_allreduce_fusion_backend": None},
        )
        self.assertEqual(_enforce_disable_allreduce_fusion(_view()), {})

        # deterministic inference disables an enabled fusion
        self.assertEqual(
            _deterministic_allreduce_fusion_disable(
                _view(
                    flashinfer_allreduce_fusion_backend="auto",
                    enable_deterministic_inference=True,
                )
            ),
            {"flashinfer_allreduce_fusion_backend": None},
        )
        self.assertEqual(
            _deterministic_allreduce_fusion_disable(
                _view(enable_deterministic_inference=True)
            ),
            {},
        )

    def test_cutedsl_prefill_backend_fill_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _cutedsl_prefill_backend_fill,
        )

        def _view(**kw):
            defaults = dict(
                attention_backend=None,
                decode_attention_backend="cutedsl_mla",
                prefill_attention_backend=None,
                kv_cache_dtype="auto",
            )
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            # decode-only cutedsl: prefill defaults to trtllm_mla
            self.assertEqual(
                _cutedsl_prefill_backend_fill(_view()),
                {"prefill_attention_backend": "trtllm_mla"},
            )
            # user-set prefill survives
            self.assertEqual(
                _cutedsl_prefill_backend_fill(_view(prefill_attention_backend="fa3")),
                {},
            )
            # cutedsl on the prefill side is rejected
            with self.assertRaises(AssertionError):
                _cutedsl_prefill_backend_fill(
                    _view(prefill_attention_backend="cutedsl_mla")
                )
            # unsupported kv dtype rejected
            with self.assertRaises(ValueError):
                _cutedsl_prefill_backend_fill(_view(kv_cache_dtype="fp8_e5m2"))
            # not a cutedsl config: nothing declared
            self.assertEqual(
                _cutedsl_prefill_backend_fill(_view(decode_attention_backend=None)),
                {},
            )
        with patch.object(overrides_module, "is_sm100_supported", return_value=False):
            with self.assertRaises(ValueError):
                _cutedsl_prefill_backend_fill(_view())

    def test_moss_vl_overrides_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import _moss_vl_overrides

        def _args(**kw):
            defaults = dict(
                attention_backend=None,
                prefill_attention_backend=None,
                decode_attention_backend=None,
            )
            defaults.update(kw)
            ns = SimpleNamespace(**defaults)
            ns.is_attention_backend_not_set = lambda: (
                ns.attention_backend is None
                and ns.prefill_attention_backend is None
                and ns.decode_attention_backend is None
            )
            ns.get_attention_backends = lambda: (
                ns.prefill_attention_backend or ns.attention_backend,
                ns.decode_attention_backend or ns.attention_backend,
            )
            return ns

        # nothing set: prefill defaults to flashinfer
        self.assertEqual(
            _moss_vl_overrides(_args(), None),
            {"prefill_attention_backend": "flashinfer"},
        )
        # compatible user choice passes with no declaration
        self.assertEqual(
            _moss_vl_overrides(_args(attention_backend="flashinfer"), None), {}
        )
        # incompatible user choice rejected
        with self.assertRaises(AssertionError):
            _moss_vl_overrides(_args(attention_backend="fa3"), None)

    def test_dsa_kv_cache_dtype_default_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _dsa_kv_cache_dtype_default,
        )

        def _view(**kw):
            hf = SimpleNamespace(architectures=["DeepseekV32ForCausalLM"])
            defaults = dict(
                kv_cache_dtype="auto",
                dsa_prefill_backend=None,
                dsa_decode_backend=None,
            )
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        with (
            patch("sglang.srt.configs.model_config.is_deepseek_dsa", return_value=True),
            patch.object(overrides_module, "is_npu", return_value=False),
            patch.object(overrides_module, "is_xpu", return_value=False),
        ):
            with patch("torch.cuda.get_device_capability", return_value=(9, 0)):
                # Hopper: auto -> bfloat16
                self.assertEqual(
                    _dsa_kv_cache_dtype_default(_view()),
                    {"kv_cache_dtype": "bfloat16"},
                )
                # alias normalization
                self.assertEqual(
                    _dsa_kv_cache_dtype_default(_view(kv_cache_dtype="bf16")),
                    {"kv_cache_dtype": "bfloat16"},
                )
                # explicit value survives (no declaration)
                self.assertEqual(
                    _dsa_kv_cache_dtype_default(_view(kv_cache_dtype="fp8_e4m3")), {}
                )
                # unsupported dtype rejected
                with self.assertRaises(AssertionError):
                    _dsa_kv_cache_dtype_default(_view(kv_cache_dtype="fp8_e5m2"))
            with patch("torch.cuda.get_device_capability", return_value=(10, 0)):
                # Blackwell: auto -> fp8
                self.assertEqual(
                    _dsa_kv_cache_dtype_default(_view()),
                    {"kv_cache_dtype": "fp8_e4m3"},
                )

    def test_deepseek_v4_kv_cache_dtype_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deepseek_v4_kv_cache_dtype,
        )

        def _view(arch="DeepseekV4ForCausalLM", **kw):
            hf = SimpleNamespace(architectures=[arch])
            defaults = dict(kv_cache_dtype="auto", device="cuda")
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        self.assertEqual(
            _deepseek_v4_kv_cache_dtype(_view()), {"kv_cache_dtype": "fp8_e4m3"}
        )
        # NPU pins bfloat16 regardless of the auto default
        self.assertEqual(
            _deepseek_v4_kv_cache_dtype(_view(device="npu")),
            {"kv_cache_dtype": "bfloat16"},
        )
        # explicit supported value survives
        self.assertEqual(
            _deepseek_v4_kv_cache_dtype(_view(kv_cache_dtype="bfloat16")), {}
        )
        with self.assertRaises(AssertionError):
            _deepseek_v4_kv_cache_dtype(_view(kv_cache_dtype="fp8_e5m2"))
        self.assertEqual(
            _deepseek_v4_kv_cache_dtype(_view(arch="LlamaForCausalLM")), {}
        )

    def test_deepseek_spec_moe_resolution_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deepseek_spec_moe_resolution,
        )
        from sglang.srt.environ import envs

        def _view(**kw):
            hf = SimpleNamespace(architectures=["DeepseekV3ForCausalLM"])
            defaults = dict(
                quantization="modelopt_fp4",
                speculative_algorithm="EAGLE",
                speculative_moe_runner_backend=None,
                speculative_moe_a2a_backend=None,
                ep_size=8,
            )
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        with patch.object(overrides_module, "is_hip", return_value=True):
            with patch.object(
                envs.SGLANG_NVFP4_CKPT_FP8_NEXTN_MOE, "get", return_value=False
            ):
                self.assertEqual(
                    _deepseek_spec_moe_resolution(_view()),
                    {
                        "speculative_moe_runner_backend": "triton",
                        "speculative_moe_a2a_backend": "none",
                    },
                )
                # guards: quantization / algorithm / both fields user-set
                self.assertEqual(
                    _deepseek_spec_moe_resolution(_view(quantization="fp8")), {}
                )
                self.assertEqual(
                    _deepseek_spec_moe_resolution(_view(speculative_algorithm=None)),
                    {},
                )
                self.assertEqual(
                    _deepseek_spec_moe_resolution(
                        _view(
                            speculative_moe_runner_backend="triton",
                            speculative_moe_a2a_backend="none",
                        )
                    ),
                    {},
                )
            with patch.object(
                envs.SGLANG_NVFP4_CKPT_FP8_NEXTN_MOE, "get", return_value=True
            ):
                self.assertEqual(
                    _deepseek_spec_moe_resolution(_view()),
                    {
                        "speculative_moe_runner_backend": "deep_gemm",
                        "speculative_moe_a2a_backend": "deepep",
                    },
                )
                with self.assertRaises(ValueError):
                    _deepseek_spec_moe_resolution(_view(ep_size=1))
        # the arm is HIP-only
        with patch.object(overrides_module, "is_hip", return_value=False):
            self.assertEqual(_deepseek_spec_moe_resolution(_view()), {})

    def test_mamba_radix_cache_resolution_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _mamba_radix_cache_resolution,
            supports_mamba_cache_extra_buffer,
        )

        def _view(arch, layer_types=None, **kw):
            hf = SimpleNamespace(architectures=[arch])
            if layer_types is not None:
                hf.layer_types = layer_types
            defaults = dict(
                disable_radix_cache=False,
                mamba_radix_cache_strategy="auto",
                disable_overlap_schedule=False,
                page_size=None,
                linear_attn_backend="triton",
            )
            defaults.update(kw)
            return ResolvedView(
                SimpleNamespace(
                    get_model_config=lambda: SimpleNamespace(hf_config=hf), **defaults
                )
            )

        # arch guard: non-mamba arch declares nothing
        self.assertEqual(_mamba_radix_cache_resolution(_view("LlamaForCausalLM")), {})
        # radix cache disabled: nothing to resolve
        self.assertEqual(
            _mamba_radix_cache_resolution(
                _view("Qwen3NextForCausalLM", disable_radix_cache=True)
            ),
            {},
        )
        # auto + overlap wanted + extra-buffer support -> extra_buffer
        self.assertEqual(
            _mamba_radix_cache_resolution(_view("Qwen3NextForCausalLM")),
            {
                "uses_mamba_radix_cache": True,
                "mamba_radix_cache_strategy": "extra_buffer",
            },
        )
        # auto + no extra-buffer support (Lfm2) -> no_buffer + overlap disable
        self.assertEqual(
            _mamba_radix_cache_resolution(_view("Lfm2ForCausalLM")),
            {
                "uses_mamba_radix_cache": True,
                "mamba_radix_cache_strategy": "no_buffer",
                "disable_overlap_schedule": True,
            },
        )
        # neither overlap nor paging wanted -> no_buffer even when supported
        declared = _mamba_radix_cache_resolution(
            _view("Qwen3NextForCausalLM", disable_overlap_schedule=True, page_size=1)
        )
        self.assertEqual(declared["mamba_radix_cache_strategy"], "no_buffer")
        self.assertIs(declared["disable_overlap_schedule"], True)
        # paging alone wants the extra buffer
        self.assertEqual(
            _mamba_radix_cache_resolution(
                _view(
                    "Qwen3NextForCausalLM", disable_overlap_schedule=True, page_size=64
                )
            )["mamba_radix_cache_strategy"],
            "extra_buffer",
        )
        # user-set strategy: only the routing marker is declared
        self.assertEqual(
            _mamba_radix_cache_resolution(
                _view(
                    "Qwen3NextForCausalLM",
                    mamba_radix_cache_strategy="extra_buffer_lazy",
                )
            ),
            {"uses_mamba_radix_cache": True},
        )
        # NemotronH routes through the pass (covered by the guard union,
        # not the branch chain — its hook invokes the handler)
        self.assertEqual(
            _mamba_radix_cache_resolution(_view("NemotronHForCausalLM")),
            {
                "uses_mamba_radix_cache": True,
                "mamba_radix_cache_strategy": "extra_buffer",
            },
        )
        # GraniteMoeHybrid is guarded on mamba layer types
        self.assertEqual(
            _mamba_radix_cache_resolution(
                _view("GraniteMoeHybridForCausalLM", layer_types=["attention"])
            ),
            {},
        )
        self.assertEqual(
            _mamba_radix_cache_resolution(
                _view("GraniteMoeHybridForCausalLM", layer_types=["mamba", "attention"])
            )["mamba_radix_cache_strategy"],
            "extra_buffer",
        )
        # extra-buffer support requires the triton linear-attn backend
        self.assertFalse(
            supports_mamba_cache_extra_buffer(
                SimpleNamespace(linear_attn_backend="fla"), "Qwen3NextForCausalLM"
            )
        )

    def test_qwen3_5_hybrid_coupled_declaration(self):
        from sglang.srt.arg_groups.overrides import _qwen3_5_hybrid_overrides

        def _args(default_backend, **kw):
            defaults = dict(
                attention_backend=None,
                _get_default_attn_backend=lambda **_: default_backend,
                use_mla_backend=lambda: False,
                get_model_config=lambda: None,
                mamba_radix_cache_strategy="auto",
                disable_radix_cache=False,
                speculative_algorithm=None,
            )
            defaults.update(kw)
            return SimpleNamespace(**defaults)

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            # radix on + no extra buffer + no spec -> page_size=1 path
            self.assertEqual(
                _qwen3_5_hybrid_overrides(_args("trtllm_mha"), None),
                {"attention_backend": "triton", "page_size": 1},
            )
            # spec decoding present -> trtllm_mha + page 64 (coupled)
            self.assertEqual(
                _qwen3_5_hybrid_overrides(
                    _args("trtllm_mha", speculative_algorithm="EAGLE"), None
                ),
                {"attention_backend": "trtllm_mha", "page_size": 64},
            )
            # user-set backend: nothing declared
            self.assertEqual(
                _qwen3_5_hybrid_overrides(
                    _args("trtllm_mha", attention_backend="fa3"), None
                ),
                {},
            )
            # the mamba pass ran before this dispatch and stashed the
            # extra-buffer strategy: the callable must see it through the
            # view (SM100 hybrid keeps trtllm_mha + page 64)
            self.assertEqual(
                _qwen3_5_hybrid_overrides(
                    _args(
                        "trtllm_mha",
                        _resolved_overrides=[
                            (
                                "_mamba_radix_cache_declarations",
                                {"mamba_radix_cache_strategy": "extra_buffer"},
                            )
                        ],
                    ),
                    None,
                ),
                {"attention_backend": "trtllm_mha", "page_size": 64},
            )
        with patch.object(overrides_module, "is_sm100_supported", return_value=False):
            self.assertEqual(_qwen3_5_hybrid_overrides(_args("fa3"), None), {})

    def test_qwen3vl_page_size(self):
        from sglang.srt.arg_groups.overrides import _qwen3vl_overrides

        with patch.object(overrides_module, "is_hip", return_value=True):
            with patch("sglang.srt.environ.envs.SGLANG_USE_AITER_UNIFIED_ATTN") as e:
                e.get.return_value = True
                self.assertEqual(
                    _qwen3vl_overrides(SimpleNamespace(page_size=None), None),
                    {"page_size": 16},
                )
                self.assertEqual(
                    _qwen3vl_overrides(SimpleNamespace(page_size=64), None), {}
                )

    def test_moe_runner_quant_constraint_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _moe_runner_backend_quant_constraints,
        )

        def _view(**kw):
            defaults = dict(quantization=None, moe_runner_backend="auto")
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            self.assertEqual(
                _moe_runner_backend_quant_constraints(
                    _view(quantization="nvfp4_online")
                ),
                {"moe_runner_backend": "flashinfer_trtllm"},
            )
            with self.assertRaises(ValueError):  # incompatible explicit backend
                _moe_runner_backend_quant_constraints(
                    _view(quantization="nvfp4_online", moe_runner_backend="triton")
                )
        self.assertEqual(
            _moe_runner_backend_quant_constraints(_view(quantization="mxfp8")),
            {"moe_runner_backend": "flashinfer_trtllm"},
        )
        with patch.object(overrides_module, "is_sm120_supported", return_value=True):
            self.assertEqual(
                _moe_runner_backend_quant_constraints(
                    _view(quantization="modelopt_fp4")
                ),
                {"moe_runner_backend": "flashinfer_cutlass"},
            )
        self.assertEqual(_moe_runner_backend_quant_constraints(_view()), {})

    def test_cutlass_moe_env_override_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _cutlass_moe_env_override,
        )

        with patch("sglang.srt.environ.envs.SGLANG_CUTLASS_MOE") as e:
            e.get.return_value = True
            self.assertEqual(
                _cutlass_moe_env_override(
                    ResolvedView(SimpleNamespace(quantization="fp8"))
                ),
                {"moe_runner_backend": "cutlass"},
            )
            with self.assertRaises(AssertionError):
                _cutlass_moe_env_override(
                    ResolvedView(SimpleNamespace(quantization=None))
                )
            e.get.return_value = False
            self.assertEqual(
                _cutlass_moe_env_override(ResolvedView(SimpleNamespace())), {}
            )

    def test_gguf_quantization_pass(self):
        from sglang.srt.arg_groups.overrides import ResolvedView, _gguf_quantization

        with patch(
            "sglang.srt.utils.hf_transformers_utils.check_gguf_file",
            return_value=True,
        ):
            self.assertEqual(
                _gguf_quantization(
                    ResolvedView(
                        SimpleNamespace(load_format="auto", model_path="x.gguf")
                    )
                ),
                {"quantization": "gguf"},
            )
            self.assertEqual(
                _gguf_quantization(
                    ResolvedView(
                        SimpleNamespace(load_format="safetensors", model_path="x")
                    )
                ),
                {},
            )

    def test_page_constraint_passes_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _fa4_page_constraint,
            _intel_xpu_page_constraint,
            _mla_backend_page_constraints,
        )

        def _view(**kw):
            defaults = dict(
                attention_backend=None,
                decode_attention_backend=None,
                prefill_attention_backend=None,
                page_size=1,
            )
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        # flashmla snaps to 64 (unconditional within the backend match)
        self.assertEqual(
            _mla_backend_page_constraints(_view(attention_backend="flashmla")),
            {"page_size": 64},
        )
        # trtllm_mla with already-valid page: no declaration
        self.assertEqual(
            _mla_backend_page_constraints(
                _view(attention_backend="trtllm_mla", page_size=32)
            ),
            {},
        )
        # chained: flashmla via decode -> 64, then trtllm_mha accepts 64
        self.assertEqual(
            _mla_backend_page_constraints(
                _view(
                    decode_attention_backend="flashmla",
                    prefill_attention_backend="trtllm_mha",
                )
            ),
            {"page_size": 64},
        )
        # no matching backend: nothing declared
        self.assertEqual(_mla_backend_page_constraints(_view()), {})

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            self.assertEqual(
                _fa4_page_constraint(
                    _view(
                        attention_backend="fa4",
                        use_mla_backend=lambda: False,
                        speculative_eagle_topk=None,
                    )
                ),
                {"page_size": 128},
            )
            self.assertEqual(
                _fa4_page_constraint(
                    _view(
                        attention_backend="fa4",
                        use_mla_backend=lambda: False,
                        speculative_eagle_topk=2,  # EAGLE topk>1 keeps default
                    )
                ),
                {},
            )

        self.assertEqual(
            _intel_xpu_page_constraint(
                _view(
                    decode_attention_backend="intel_xpu",
                    use_mla_backend=lambda: False,
                )
            ),
            {"page_size": 128},
        )
        self.assertEqual(
            _intel_xpu_page_constraint(
                _view(
                    decode_attention_backend="intel_xpu",
                    use_mla_backend=lambda: True,
                    page_size=16,  # MLA decode accepts 16
                )
            ),
            {},
        )

    def test_monolith_attention_families_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import (
            _falcon_h1_jet_overrides,
            _gemma4_overrides,
            _glm4_moe_overrides,
            _granite_moe_hybrid_overrides,
            _lfm2_overrides,
            _llama4_overrides,
            _minicpm_v4_6_overrides,
        )

        def _args(**kw):
            defaults = dict(
                device="cuda",
                attention_backend=None,
                is_attention_backend_not_set=lambda: True,
                # keep the (now-absorbed) quant/moe blocks inert so these
                # assertions stay attention-only
                moe_runner_backend="triton",
                quantization=None,
            )
            defaults.update(kw)
            return SimpleNamespace(**defaults)

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            self.assertEqual(
                _llama4_overrides(_args(), None), {"attention_backend": "trtllm_mha"}
            )
            self.assertEqual(_llama4_overrides(_args(device="cpu"), None), {})
            self.assertEqual(
                _llama4_overrides(_args(attention_backend="fa3"), None), {}
            )
            self.assertEqual(
                _gemma4_overrides(_args(), None), {"attention_backend": "trtllm_mha"}
            )
            self.assertEqual(
                _minicpm_v4_6_overrides(_args(), None),
                {"attention_backend": "triton"},
            )
            self.assertEqual(
                _falcon_h1_jet_overrides(_args(), None),
                {"attention_backend": "triton"},
            )
            self.assertEqual(
                _granite_moe_hybrid_overrides(
                    _args(), SimpleNamespace(layer_types=["mamba", "attention"])
                ),
                {"attention_backend": "flashinfer"},
            )
            self.assertEqual(
                _granite_moe_hybrid_overrides(
                    _args(), SimpleNamespace(layer_types=["attention"])
                ),
                {},
            )
            self.assertEqual(
                _lfm2_overrides(_args(), None), {"attention_backend": "flashinfer"}
            )
        with patch.object(overrides_module, "is_sm100_supported", return_value=False):
            self.assertEqual(_minicpm_v4_6_overrides(_args(), None), {})
            with patch.object(overrides_module, "is_sm90_supported", return_value=True):
                self.assertEqual(
                    _llama4_overrides(_args(), None), {"attention_backend": "fa3"}
                )
            self.assertEqual(
                _gemma4_overrides(_args(), None), {"attention_backend": "triton"}
            )
        # Glm4Moe: unconditional tf32 declaration + (sm100) quant/moe absorption
        with patch.object(overrides_module, "is_sm100_supported", return_value=False):
            self.assertEqual(
                _glm4_moe_overrides(None, None), {"enable_tf32_matmul": True}
            )
        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            self.assertEqual(
                _glm4_moe_overrides(
                    SimpleNamespace(
                        quantization=None,
                        _quantization_explicitly_unset=False,
                        moe_a2a_backend="none",
                        moe_runner_backend="auto",
                    ),
                    SimpleNamespace(
                        quantization_config={"quant_method": "modelopt_fp4"}
                    ),
                ),
                {
                    "quantization": "modelopt_fp4",
                    "moe_runner_backend": "flashinfer_trtllm",
                    "enable_tf32_matmul": True,
                },
            )

    def test_deepseek_moe_quant_slot_pass(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _deepseek_moe_quant_resolution,
        )

        def _view(arch="DeepseekV32ForCausalLM", quant_cfg=None, **kw):
            defaults = dict(
                quantization=None,
                _quantization_explicitly_unset=False,
                moe_a2a_backend="none",
                moe_runner_backend="auto",
                get_model_config=lambda: SimpleNamespace(
                    hf_config=SimpleNamespace(
                        architectures=[arch], quantization_config=quant_cfg
                    )
                ),
            )
            defaults.update(kw)
            return ResolvedView(SimpleNamespace(**defaults))

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            with patch.object(
                overrides_module, "get_quantization_config", return_value="fp8"
            ):
                # config-declared quant: detected + moe runner
                self.assertEqual(
                    _deepseek_moe_quant_resolution(_view()),
                    {
                        "quantization": "fp8",
                        "moe_runner_backend": "flashinfer_trtllm",
                    },
                )
            # non-deepseek arch guard (end-state list execution safety)
            self.assertEqual(
                _deepseek_moe_quant_resolution(_view(arch="LlamaForCausalLM")), {}
            )
        with patch.object(overrides_module, "is_sm100_supported", return_value=False):
            self.assertEqual(_deepseek_moe_quant_resolution(_view()), {})

    def test_data_parallelism_and_a2a_passes(self):
        from sglang.srt.arg_groups.overrides import (
            ResolvedView,
            _a2a_backend_overrides,
            _a2a_ep_size,
            _data_parallelism_defaults,
        )

        self.assertEqual(
            _data_parallelism_defaults(ResolvedView(SimpleNamespace(dp_size=1))),
            {"enable_dp_attention": False, "enable_dp_lm_head": False},
        )
        self.assertEqual(
            _data_parallelism_defaults(ResolvedView(SimpleNamespace(dp_size=2))), {}
        )

        with patch("sglang.srt.environ.envs.SGLANG_OPT_USE_DEEPGEMM_MEGA_MOE") as e:
            e.get.return_value = False
            self.assertEqual(
                _a2a_backend_overrides(
                    ResolvedView(
                        SimpleNamespace(
                            enable_deepep_waterfill=True, moe_a2a_backend="none"
                        )
                    )
                ),
                {"moe_a2a_backend": "deepep"},
            )
            e.get.return_value = True
            # megamoe env wins over the waterfill override (chained, last write)
            self.assertEqual(
                _a2a_backend_overrides(
                    ResolvedView(
                        SimpleNamespace(
                            enable_deepep_waterfill=True, moe_a2a_backend="none"
                        )
                    )
                ),
                {"moe_a2a_backend": "megamoe"},
            )

        self.assertEqual(
            _a2a_ep_size(
                ResolvedView(SimpleNamespace(moe_a2a_backend="deepep", tp_size=8))
            ),
            {"ep_size": 8},
        )
        self.assertEqual(
            _a2a_ep_size(
                ResolvedView(SimpleNamespace(moe_a2a_backend="none", tp_size=8))
            ),
            {},
        )

    def test_deepseek_family_order_safe_declarations(self):
        from sglang.srt.arg_groups.overrides import _deepseek_family_overrides

        def _args(**kw):
            defaults = dict(
                is_attention_backend_not_set=lambda: True,
                attention_backend=None,
                prefill_attention_backend=None,
                decode_attention_backend=None,
                enable_prefill_cp=False,
            )
            defaults.update(kw)
            return SimpleNamespace(**defaults)

        # DSA path on CUDA: dsa fill + page 64
        with patch(
            "sglang.srt.configs.model_config.is_deepseek_dsa", return_value=True
        ):
            with patch.object(overrides_module, "is_npu", return_value=False):
                with patch.object(overrides_module, "is_xpu", return_value=False):
                    with patch.object(overrides_module, "is_hip", return_value=False):
                        self.assertEqual(
                            _deepseek_family_overrides(_args(), None),
                            {"attention_backend": "dsa", "page_size": 64},
                        )
                    # HIP without the preshuffle path: page 1
                    with patch.object(overrides_module, "is_hip", return_value=True):
                        with patch(
                            "sglang.srt.layers.attention.dsa.utils.aiter_can_use_preshuffle_paged_mqa",
                            return_value=False,
                        ):
                            self.assertEqual(
                                _deepseek_family_overrides(_args(), None),
                                {"attention_backend": "dsa", "page_size": 1},
                            )
        # DSA CP (zigzag): the coupled parallel-field declaration
        with patch(
            "sglang.srt.configs.model_config.is_deepseek_dsa", return_value=True
        ):
            with patch.object(overrides_module, "is_npu", return_value=False):
                with patch.object(overrides_module, "is_xpu", return_value=False):
                    with patch.object(overrides_module, "is_hip", return_value=False):
                        result = _deepseek_family_overrides(
                            _args(
                                enable_prefill_cp=True,
                                cp_strategy="zigzag",
                                tp_size=8,
                                dp_size=1,
                                ep_size=1,
                                moe_a2a_backend="none",
                                kv_cache_dtype="auto",
                            ),
                            None,
                        )
                        self.assertEqual(
                            result,
                            {
                                "attention_backend": "dsa",
                                "page_size": 64,
                                "enable_dp_attention": True,
                                "moe_dense_tp_size": 1,
                                "moe_a2a_backend": "deepep",
                                "ep_size": 8,
                                "attn_cp_size": 8,
                            },
                        )
                        # interleave CP with dp>1 must assert
                        with self.assertRaises(AssertionError):
                            _deepseek_family_overrides(
                                _args(
                                    enable_prefill_cp=True,
                                    cp_strategy="interleave",
                                    tp_size=8,
                                    dp_size=2,
                                ),
                                None,
                            )

        # MLA path on sm100: trtllm_mla fill (all three backends unset)
        with patch(
            "sglang.srt.configs.model_config.is_deepseek_dsa", return_value=False
        ):
            with patch.object(
                overrides_module, "is_sm100_supported", return_value=True
            ):
                self.assertEqual(
                    _deepseek_family_overrides(_args(), None),
                    {"attention_backend": "trtllm_mla"},
                )
                self.assertEqual(
                    _deepseek_family_overrides(
                        _args(decode_attention_backend="fa3"), None
                    ),
                    {},
                )
            with patch.object(
                overrides_module, "is_sm100_supported", return_value=False
            ):
                self.assertEqual(_deepseek_family_overrides(_args(), None), {})

    def test_qwen3_moe_family_quant_absorption(self):
        from sglang.srt.arg_groups.overrides import _qwen3_moe_family_overrides

        with patch.object(overrides_module, "is_sm100_supported", return_value=True):
            with patch.object(
                overrides_module, "get_quantization_config", return_value="fp8"
            ):
                self.assertEqual(
                    _qwen3_moe_family_overrides(
                        SimpleNamespace(
                            quantization=None,
                            _quantization_explicitly_unset=False,
                            moe_a2a_backend="none",
                            moe_runner_backend="auto",
                        ),
                        SimpleNamespace(architectures=["Qwen3MoeForCausalLM"]),
                    ),
                    {
                        "quantization": "fp8",
                        "moe_runner_backend": "flashinfer_trtllm",
                    },
                )
        with patch.object(overrides_module, "is_sm100_supported", return_value=False):
            self.assertEqual(_qwen3_moe_family_overrides(None, None), {})

    def test_step3p_declarations_at_callable_level(self):
        from sglang.srt.arg_groups.overrides import _step3p_overrides

        def _args(**kw):
            defaults = dict(
                speculative_algorithm=None,
                enable_hierarchical_cache=False,
                is_attention_backend_not_set=lambda: False,
            )
            defaults.update(kw)
            return SimpleNamespace(**defaults)

        self.assertEqual(
            _step3p_overrides(_args(speculative_algorithm="EAGLE"), None),
            {"enable_multi_layer_eagle": True},
        )
        self.assertEqual(
            _step3p_overrides(_args(enable_hierarchical_cache=True), None),
            {"swa_full_tokens_ratio": 1.0, "disable_hybrid_swa_memory": True},
        )
        self.assertEqual(_step3p_overrides(_args(), None), {})


class TestDeclarationValidation(CustomTestCase):
    def test_declarations_never_mutate_server_args(self):
        args = _FakeArgs()
        declarations = [("src", {"resolved_by_model": "dsv4", "also_resolved": 7})]
        validate_declarations(args, declarations)
        # validation is a pure whitelist check: the fields stay untouched
        self.assertEqual(args.resolved_by_model, _FakeArgs.resolved_by_model)
        self.assertEqual(args.also_resolved, _FakeArgs.also_resolved)

    def test_validation_rejects_unknown_fields(self):
        args = _FakeArgs()
        with self.assertRaises(ValueError):
            validate_declarations(args, [("src", {"nope": 1})])


if __name__ == "__main__":
    unittest.main()