[diffusion] Restrict request-level quality to two validated tiers: lossless (default) and high (#33453)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

python/sglang/multimodal_gen/configs/pipeline_configs/minimax_h3.py

@@ -84,7 +84,8 @@ class MiniMaxH3PipelineConfig(PipelineConfig):
         return getattr(value, "value", value)
 
     def validate_quality_deployment(self, server_args) -> None:
-        """Fail closed unless the resident server matches the measured profile."""
+        """Fail closed unless the resident server matches the deployment
+        audited for quality="high"."""
 
         attention_backend = self._server_arg_value(server_args.attention_backend)
         attention_backend = (
@@ -161,7 +162,7 @@ class MiniMaxH3PipelineConfig(PipelineConfig):
             }
         if mismatches:
             raise ValueError(
-                "MiniMax-H3 approximate quality profiles are validated only for "
+                'MiniMax-H3 quality="high" is validated only for '
                 f"the strict 4xH200 fl2va deployment; mismatches: {mismatches}"
             )
 

python/sglang/multimodal_gen/configs/sample/sampling_params.py

@@ -51,6 +51,12 @@ def generate_request_id() -> str:
     return str(uuid.uuid4())
 
 
+# Validated request-level quality levels. "lossless" is the exact reference
+# path (bit-exact against the CI golden outputs); "high" opts into validated
+# accelerated paths whose quality is guaranteed but not bit-exact.
+QUALITY_LEVELS: tuple[str, ...] = ("lossless", "high")
+
+
 def _sanitize_filename(name: str, replacement: str = "_", max_length: int = 150) -> str:
     """Create a filesystem- and ffmpeg-friendly filename.
 
@@ -123,9 +129,20 @@ class SamplingParams:
     )
     output_quality: str | None = "default"
     output_compression: int | None = None
-    # Model-owned, request-scoped approximate acceleration profile. Models
-    # that support it must validate the deployment and workload explicitly.
-    # It intentionally participates in the dynamic-batch signature.
+    # Model-owned, request-scoped quality level.
+    #
+    # - "lossless" (default): the exact reference path. Output is expected to
+    #   be bit-identical to the HF reference implementation and to pass the
+    #   CI golden/ground-truth comparisons.
+    # - "high": opt into validated accelerated paths. Quality stays
+    #   guaranteed (the intent is to back every such path with mathematical
+    #   acceptance thresholds, e.g. PSNR > 25 against the reference), but
+    #   the output is no longer bit-exact versus the HF reference or the CI
+    #   ground truth.
+    #
+    # Models that support "high" must validate the deployment and workload
+    # explicitly. It intentionally participates in the dynamic-batch
+    # signature.
     quality: str = "lossless"
 
     # Frame interpolation
@@ -408,9 +425,10 @@ class SamplingParams:
                 f"prompt_path must be a txt file, got {self.prompt_path!r}"
             )
 
-        if not isinstance(self.quality, str) or not self.quality.strip():
+        if self.quality not in QUALITY_LEVELS:
             raise ValueError(
-                f"quality must be a non-empty string, got {self.quality!r}"
+                f"quality must be one of {list(QUALITY_LEVELS)}, "
+                f"got {self.quality!r}"
             )
 
         # These are always required to be sane regardless of pipeline.
@@ -932,9 +950,14 @@ class SamplingParams:
         add_argument(
             "--quality",
             type=str,
+            choices=list(QUALITY_LEVELS),
             help=(
-                "Select a model-owned quality/performance profile. "
-                "Support and validated deployment constraints are model-specific."
+                "Request-level quality: 'lossless' (default) keeps the exact "
+                "reference path, bit-exact against the reference "
+                "implementation; 'high' opts into the model-owned validated "
+                "accelerated path, whose quality stays guaranteed but is not "
+                "bit-exact. Support and validated deployment constraints are "
+                "model-specific."
             ),
         )
         add_argument(

python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/minimax_h3/constants.py

@@ -27,11 +27,10 @@ MINIMAX_H3_MAX_DURATION_SECONDS = 15.0
 # The distilled checkpoint has exactly one positive denoise branch.
 MINIMAX_H3_DEFAULT_BRANCHES: tuple = ({"name": "cond_1"},)
 
-# Audited 4xH200 T2VA profiles. The tuple is
+# Audited 4xH200 T2VA Cache-DiT parameters for quality="high":
 # (warmup steps, residual-difference threshold, max consecutive cached steps).
-MINIMAX_H3_QUALITY_PROFILES: dict[str, tuple[int, float, int] | None] = {
-    "lossless": None,
-    "high": (4, 0.04, 1),
-    "medium": (4, 0.12, 3),
-    "low": (4, 0.24, 3),
-}
+# Measured SSIM 0.931 / PSNR 28.16 dB against quality="lossless" on the
+# validated workload. quality="lossless" (the default) uses no Cache-DiT
+# configuration at all. Process-wide SGLANG_CACHE_DIT_* environment controls
+# remain available for manual experiments and are independent of this field.
+MINIMAX_H3_HIGH_QUALITY_CACHE_DIT_CONFIG: tuple[int, float, int] = (4, 0.04, 1)

python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/minimax_h3/release_metadata.py

@@ -7,11 +7,9 @@ import math
 from dataclasses import dataclass
 from typing import Any, Mapping
 
+from sglang.multimodal_gen.configs.sample.sampling_params import QUALITY_LEVELS
 from sglang.multimodal_gen.runtime.pipelines_core.schedule_batch import Req
 from sglang.multimodal_gen.runtime.pipelines_core.stages.base import PipelineStage
-from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.constants import (
-    MINIMAX_H3_QUALITY_PROFILES,
-)
 from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.resolved_plan import (
     minimax_h3_plan_from_batch,
 )
@@ -150,25 +148,23 @@ class MiniMaxH3PartitionAdmissionStage(PipelineStage):
             raise ValueError("MiniMax H3 request task must be a non-empty string")
         self.metadata.canonical_task(task)
         quality = getattr(batch.sampling_params, "quality", "lossless")
-        if quality not in MINIMAX_H3_QUALITY_PROFILES:
+        if quality not in QUALITY_LEVELS:
             raise ValueError(
-                f"unsupported MiniMax-H3 quality profile {quality!r}; supported: "
-                f"{list(MINIMAX_H3_QUALITY_PROFILES)}"
+                f"quality must be one of {list(QUALITY_LEVELS)}, got {quality!r}"
             )
-        approximate = quality != "lossless"
+        high_quality = quality == "high"
         attention_backend = str(server_args.attention_backend or "").strip().lower()
         if attention_backend == "sage_attn" and not batch.is_warmup:
             raise ValueError(
                 "MiniMax-H3 does not support SageAttention: the current packed "
                 "varlen path does not preserve model output"
             )
-        if approximate and not batch.is_warmup:
+        if high_quality and not batch.is_warmup:
             server_args.pipeline_config.validate_quality_deployment(server_args)
             plan = minimax_h3_plan_from_batch(batch)
             if plan is None:
                 raise ValueError(
-                    "MiniMax-H3 approximate quality profiles require a resolved "
-                    "request plan"
+                    'MiniMax-H3 quality="high" requires a resolved request plan'
                 )
             shape = plan.shape
             actual = {
@@ -212,7 +208,7 @@ class MiniMaxH3PartitionAdmissionStage(PipelineStage):
             )
             if not exact or not shifts:
                 raise ValueError(
-                    "MiniMax-H3 approximate quality profiles are validated only for "
+                    'MiniMax-H3 quality="high" is validated only for '
                     f"{_MINIMAX_H3_QUALITY_WORKLOAD}; got {actual}"
                 )
         return batch

python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/minimax_h3/stages/denoising.py

@@ -24,7 +24,7 @@ from sglang.multimodal_gen.runtime.pipelines_core.stages.denoising import (
     DenoisingStage,
 )
 from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.constants import (
-    MINIMAX_H3_QUALITY_PROFILES,
+    MINIMAX_H3_HIGH_QUALITY_CACHE_DIT_CONFIG,
 )
 from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.task_profiles import (
     MINIMAX_H3_FL2VA_KEYFRAME_SIGNATURES,
@@ -387,7 +387,7 @@ class MiniMaxH3DenoisingStage(DenoisingStage):
             scheduler=None,
             pipeline=pipeline,
         )
-        self._minimax_h3_quality_profile = "lossless"
+        self._minimax_h3_quality = "lossless"
         self._minimax_h3_cache_mode: str | None = None
 
     def _owns_compile_warmup_lifecycle(self) -> bool:
@@ -395,7 +395,7 @@ class MiniMaxH3DenoisingStage(DenoisingStage):
 
     def _cache_dit_requested(self) -> bool:
         return (
-            getattr(self, "_minimax_h3_quality_profile", "lossless") != "lossless"
+            getattr(self, "_minimax_h3_quality", "lossless") == "high"
             or super()._cache_dit_requested()
         )
 
@@ -403,19 +403,15 @@ class MiniMaxH3DenoisingStage(DenoisingStage):
         self, num_inference_steps: int | tuple[int, int], batch: Req
     ) -> None:
         quality = getattr(batch.sampling_params, "quality", "lossless")
-        if quality not in MINIMAX_H3_QUALITY_PROFILES:
-            raise ValueError(f"unsupported MiniMax-H3 quality profile {quality!r}")
         explicit_fields = getattr(batch.sampling_params, "_explicit_fields", ())
         generic_requested = (
             super()._cache_dit_requested() and "quality" not in explicit_fields
         )
         desired_mode = (
-            quality
-            if quality != "lossless"
-            else ("generic" if generic_requested else None)
+            "high" if quality == "high" else ("generic" if generic_requested else None)
         )
         current_mode = getattr(self, "_minimax_h3_cache_mode", None)
-        self._minimax_h3_quality_profile = quality
+        self._minimax_h3_quality = quality
 
         # H3 is monolithic-only, and the scheduler executes one worker batch at
         # a time. Combined with `quality` in the dynamic-batch signature, this
@@ -435,7 +431,7 @@ class MiniMaxH3DenoisingStage(DenoisingStage):
     def _cache_dit_scm_masks(
         self, primary_num_steps: int, secondary_num_steps: int | None = None
     ) -> tuple[str, str, list[int] | None, list[int] | None]:
-        if getattr(self, "_minimax_h3_quality_profile", "lossless") != "lossless":
+        if getattr(self, "_minimax_h3_quality", "lossless") == "high":
             return "none", "dynamic", None, None
         return super()._cache_dit_scm_masks(primary_num_steps, secondary_num_steps)
 
@@ -447,16 +443,14 @@ class MiniMaxH3DenoisingStage(DenoisingStage):
         *,
         secondary: bool = False,
     ) -> CacheDitConfig:
-        quality = getattr(self, "_minimax_h3_quality_profile", "lossless")
-        profile = MINIMAX_H3_QUALITY_PROFILES[quality]
-        if profile is None or secondary:
+        if secondary or getattr(self, "_minimax_h3_quality", "lossless") != "high":
             return super()._build_cache_dit_config(
                 num_inference_steps,
                 steps_computation_mask,
                 scm_policy,
                 secondary=secondary,
             )
-        warmup, threshold, max_cached = profile
+        warmup, threshold, max_cached = MINIMAX_H3_HIGH_QUALITY_CACHE_DIT_CONFIG
         return CacheDitConfig(
             enabled=True,
             Fn_compute_blocks=1,

python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/minimax_h3/video_adapter.py

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any
 from sglang.multimodal_gen.configs.pipeline_configs.minimax_h3 import (
     MiniMaxH3PipelineConfig,
 )
+from sglang.multimodal_gen.configs.sample.sampling_params import QUALITY_LEVELS
 from sglang.multimodal_gen.runtime.entrypoints.openai.protocol import (
     VideoGenerationsRequest,
 )
@@ -104,12 +105,14 @@ class MiniMaxH3VideoModelAdapter:
         request: VideoGenerationsRequest,
         name: str,
     ) -> str | None:
-        value = _parse_extra_value(_extra_value(request, name))
+        value = _extra_value(request, name)
         if value is None:
             return None
-        if not isinstance(value, str) or not value.strip():
-            raise ValueError(f"{name} must be a non-empty string")
-        return value.strip().lower()
+        if value not in QUALITY_LEVELS:
+            raise ValueError(
+                f"{name} must be one of {list(QUALITY_LEVELS)}, got {value!r}"
+            )
+        return value
 
     @staticmethod
     def _reject_retired_cfg_fields(kwargs: dict[str, Any]) -> None:

python/sglang/multimodal_gen/test/unit/test_minimax_h3_admission.py

@@ -208,6 +208,21 @@ def test_video_adapter_lowers_only_native_fields_and_rejects_cfg():
         )
 
 
+@pytest.mark.parametrize("bad_quality", ["ultra", "draft", "", 1])
+def test_video_adapter_rejects_invalid_quality(bad_quality):
+    request = VideoGenerationsRequest(
+        prompt="contract",
+        task="t2va",
+        conditions=[],
+        target=TARGET,
+        quality=bad_quality,
+    )
+    with pytest.raises(ValueError, match="quality must be one of"):
+        MiniMaxH3SamplingParams.lower_video_request_kwargs(
+            request, {"prompt": request.prompt, "seed": request.seed}
+        )
+
+
 class _HopperCapability:
     def to_int(self) -> int:
         return 90
@@ -289,7 +304,7 @@ def test_quality_admission_fails_closed_outside_validated_request():
     with pytest.raises(ValueError, match="does not support SageAttention"):
         stage.forward(batch, server_args)
 
-    batch.sampling_params.quality = "unsupported"
+    batch.sampling_params.quality = "ultra"
     server_args.attention_backend = None
-    with pytest.raises(ValueError, match="unsupported MiniMax-H3 quality profile"):
+    with pytest.raises(ValueError, match="quality must be one of"):
         stage.forward(batch, server_args)

python/sglang/multimodal_gen/test/unit/test_sampling_params.py

@@ -41,12 +41,17 @@ class TestSamplingParamsValidate(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, r"num_outputs_per_prompt"):
             SamplingParams(num_outputs_per_prompt=0)
 
-    def test_quality_must_be_a_non_empty_profile_name(self):
+    def test_quality_defaults_to_lossless(self):
+        self.assertEqual(SamplingParams().quality, "lossless")
+
+    def test_quality_accepts_the_two_validated_levels(self):
+        self.assertEqual(SamplingParams(quality="lossless").quality, "lossless")
         self.assertEqual(SamplingParams(quality="high").quality, "high")
-        with self.assertRaisesRegex(ValueError, r"quality must be a non-empty string"):
-            SamplingParams(quality="")
-        with self.assertRaisesRegex(ValueError, r"quality must be a non-empty string"):
-            SamplingParams(quality=True)  # type: ignore[arg-type]
+
+    def test_quality_rejects_invalid_values(self):
+        for bad in ("ultra", "draft", "fast", "", True, 1):
+            with self.assertRaisesRegex(ValueError, r"quality must be one of"):
+                SamplingParams(quality=bad)  # type: ignore[arg-type]
 
     def test_seed_accepts_int_or_non_empty_int_list(self):
         self.assertEqual(SamplingParams(seed=7).seed, 7)
@@ -235,7 +240,7 @@ class TestSamplingParamsCliArgs(unittest.TestCase):
     def test_quality_is_request_scoped_cli_arg(self):
         self.assertNotIn("quality", self._parse_cli_kwargs([]))
         self.assertEqual(
-            self._parse_cli_kwargs(["--quality", "medium"])["quality"], "medium"
+            self._parse_cli_kwargs(["--quality", "high"])["quality"], "high"
         )
 
     def test_qwen_image_cli_path_preserves_model_defaults(self):