feat(vllm-omni): add MiniMax-H3 request adapter

components/src/dynamo/common/protocols/video_protocol.py

@@ -8,10 +8,17 @@
 """
 # TODO: Replace these Pydantic models with Python bindings to the Rust protocol types once PyO3 bindings are available.
 
-from typing import Literal, Optional
+from typing import Literal, Optional, Union
 
 from pydantic import BaseModel, model_validator
 
+H3Task = Literal["t2va", "fl2va", "ref2va"]
+
+
+def is_minimax_h3_model_name(model: str) -> bool:
+    """Return whether a public model identifier names MiniMax-H3."""
+    return "minimax-h3" in model.lower().replace("_", "-")
+
 
 class VideoInputReference(BaseModel):
     """Typed conditioning input for video generation."""
@@ -56,6 +63,51 @@ class VideoNvExt(BaseModel):
     guidance_scale_2: Optional[float] = None
     """CFG scale for the low-noise expert (vLLM-Omni I2V dual-guidance)."""
 
+    task: Optional[H3Task] = None
+    """MiniMax-H3 task routed to its FL2VA or Ref2VA transformer."""
+
+    duration: Optional[float] = None
+    """Requested MiniMax-H3 duration in seconds (4 through 15)."""
+
+    flow_shift: Optional[float] = None
+    """MiniMax-H3 video sigma shift."""
+
+    audio_flow_shift: Optional[float] = None
+    """MiniMax-H3 audio sigma shift."""
+
+    aspect_ratio: Optional[str] = None
+    """MiniMax-H3 output aspect ratio."""
+
+    short_edge: Optional[int] = None
+    """MiniMax-H3 output canvas short edge."""
+
+    frame_indices: Optional[list[int]] = None
+    """FL2VA keyframe positions: [0], [-1], or [0, -1]."""
+
+    start_time_seconds: Optional[Union[float, list[float]]] = None
+    """Start offset for one reference video, or one offset per video."""
+
+    num_outputs_per_prompt: Optional[int] = None
+    """Number of generated videos (MiniMax-H3 supports 1 through 10)."""
+
+    quality: Optional[Literal["lossless", "high"]] = None
+    """MiniMax-H3 request-scoped quality policy."""
+
+    @model_validator(mode="after")
+    def validate_h3_fields(self) -> "VideoNvExt":
+        if self.duration is not None and not 4 <= self.duration <= 15:
+            raise ValueError("duration must be between 4 and 15 seconds")
+        if self.num_outputs_per_prompt is not None and not (
+            1 <= self.num_outputs_per_prompt <= 10
+        ):
+            raise ValueError("num_outputs_per_prompt must be between 1 and 10")
+        if self.task is not None and self.fps is not None and self.fps != 24:
+            raise ValueError("MiniMax-H3 fps is fixed at 24")
+        if self.task == "fl2va" and self.frame_indices is not None:
+            if self.frame_indices not in ([0], [-1], [0, -1]):
+                raise ValueError("FL2VA frame_indices must be [0], [-1], or [0, -1]")
+        return self
+
 
 class NvCreateVideoRequest(BaseModel):
     """Request for video generation (/v1/videos endpoint).
@@ -108,8 +160,75 @@ def validate_input_references(self) -> "NvCreateVideoRequest":
             )
         if self.input_references is not None and not self.input_references:
             raise ValueError("input_references must not be empty")
+        if self.input_references is not None and len(self.input_references) > 12:
+            raise ValueError("input_references accepts at most 12 references")
+
+        task = self.nvext.task if self.nvext is not None else None
+        if task is not None or is_minimax_h3_model_name(self.model):
+            self.validate_h3_reference_contract(task or self.infer_h3_task())
         return self
 
+    def infer_h3_task(self) -> H3Task:
+        """Match MiniMax-H3's task inference from the supplied references."""
+        if self.nvext is not None and self.nvext.task is not None:
+            return self.nvext.task
+        reference_types = {reference.type for reference in self.input_references or []}
+        if reference_types.intersection({"video", "audio"}):
+            return "ref2va"
+        if self.input_reference is not None or "image" in reference_types:
+            return "fl2va"
+        return "t2va"
+
+    def validate_h3_reference_contract(self, task: H3Task) -> None:
+        """Validate controls and references for a resolved MiniMax-H3 task."""
+        nvext = self.nvext or VideoNvExt()
+        if nvext.fps is not None and nvext.fps != 24:
+            raise ValueError("MiniMax-H3 fps is fixed at 24")
+        if task == "fl2va" and nvext.frame_indices is not None:
+            if nvext.frame_indices not in ([0], [-1], [0, -1]):
+                raise ValueError("FL2VA frame_indices must be [0], [-1], or [0, -1]")
+
+        counts = {kind: 0 for kind in ("image", "video", "audio")}
+        if self.input_reference is not None:
+            counts["image"] = 1
+        for reference in self.input_references or []:
+            counts[reference.type] += 1
+        total = sum(counts.values())
+
+        if task == "t2va" and total:
+            raise ValueError("t2va does not accept input references")
+        if task == "fl2va":
+            if counts["video"] or counts["audio"] or not counts["image"]:
+                raise ValueError("fl2va accepts only one or two image references")
+            if counts["image"] > 2:
+                raise ValueError("fl2va accepts at most two image references")
+            if (
+                nvext.frame_indices is not None
+                and len(nvext.frame_indices) != counts["image"]
+            ):
+                raise ValueError("fl2va requires one frame index per image reference")
+        if task == "ref2va":
+            if not counts["image"] and not counts["video"]:
+                raise ValueError(
+                    "ref2va requires at least one image or video reference"
+                )
+            if counts["image"] > 9:
+                raise ValueError("ref2va accepts at most 9 image references")
+            if counts["video"] > 3:
+                raise ValueError("ref2va accepts at most 3 video references")
+            if counts["audio"] > 3:
+                raise ValueError("ref2va accepts at most 3 audio references")
+
+        start_times = nvext.start_time_seconds
+        if isinstance(start_times, list) and len(start_times) != counts["video"]:
+            raise ValueError(
+                "start_time_seconds requires one value per video reference"
+            )
+        if isinstance(start_times, float) and counts["video"] != 1:
+            raise ValueError(
+                "scalar start_time_seconds requires exactly one video reference"
+            )
+
 
 class VideoData(BaseModel):
     """Video data in response.

components/src/dynamo/common/tests/test_video_protocol.py

@@ -68,6 +68,102 @@ def test_video_request_rejects_invalid_reference_combinations(input_references):
         NvCreateVideoRequest(prompt="cat", model="video-model", **kwargs)
 
 
+def test_video_request_rejects_more_than_twelve_references():
+    with pytest.raises(ValueError, match="at most 12"):
+        NvCreateVideoRequest(
+            prompt="cat",
+            model="video-model",
+            input_references=[
+                {"type": "image", "source": f"https://example.com/{index}.png"}
+                for index in range(13)
+            ],
+        )
+
+
+@pytest.mark.parametrize(
+    ("task", "references", "nvext", "message"),
+    [
+        (
+            "t2va",
+            [{"type": "image", "source": "https://example.com/cat.png"}],
+            {},
+            "does not accept",
+        ),
+        (
+            "fl2va",
+            [{"type": "audio", "source": "https://example.com/cat.wav"}],
+            {},
+            "only one or two image",
+        ),
+        (
+            "fl2va",
+            [{"type": "image", "source": "https://example.com/cat.png"}],
+            {"frame_indices": [0, -1]},
+            "one frame index per image",
+        ),
+        (
+            "ref2va",
+            [{"type": "audio", "source": "https://example.com/cat.wav"}],
+            {},
+            "at least one image or video",
+        ),
+        (
+            "ref2va",
+            [{"type": "video", "source": "https://example.com/cat.mp4"}],
+            {"start_time_seconds": [0.0, 1.0]},
+            "one value per video",
+        ),
+    ],
+)
+def test_video_request_rejects_invalid_h3_reference_contract(
+    task, references, nvext, message
+):
+    with pytest.raises(ValueError, match=message):
+        NvCreateVideoRequest(
+            prompt="cat",
+            model="MiniMaxAI/MiniMax-H3",
+            input_references=references,
+            nvext={"task": task, **nvext},
+        )
+
+
+@pytest.mark.parametrize(
+    "references",
+    [
+        [{"type": "audio", "source": "https://example.com/cat.wav"}],
+        [
+            {"type": "image", "source": f"https://example.com/{index}.png"}
+            for index in range(3)
+        ],
+        [
+            {"type": "video", "source": f"https://example.com/{index}.mp4"}
+            for index in range(4)
+        ],
+    ],
+)
+def test_video_request_rejects_invalid_taskless_h3_reference_contract(references):
+    with pytest.raises(ValueError):
+        NvCreateVideoRequest(
+            prompt="cat",
+            model="MiniMaxAI/MiniMax-H3",
+            input_references=references,
+        )
+
+
+@pytest.mark.parametrize(
+    "nvext",
+    [
+        {"task": "t2va", "fps": 16},
+        {"task": "fl2va", "frame_indices": [1]},
+        {"task": "ref2va", "duration": 3},
+        {"task": "ref2va", "num_outputs_per_prompt": 11},
+    ],
+)
+def test_video_request_rejects_invalid_h3_controls(nvext):
+    with pytest.raises(ValueError):
+        NvCreateVideoRequest(prompt="cat", model="MiniMaxAI/MiniMax-H3", nvext=nvext)
+
+
 def test_video_response_wire_shape():
     response = NvVideosResponse(
         id="r1",

components/src/dynamo/vllm/omni/args.py

@@ -49,6 +49,12 @@ class OmniDiffusionKwargs:
     cache_config: Optional[str] = None
     enable_cache_dit_summary: bool = False
     enable_cpu_offload: bool = False
+    task_type: Optional[str] = None
+    diffusion_attention_backend: Optional[str] = None
+    diffusion_attention_config: Optional[str] = None
+    enable_distributed_layerwise_offload: bool = False
+    dlo_use_allgather: bool = True
+    dlo_resident_layers: int = 0
     enforce_eager: bool = False
 
 
@@ -184,6 +190,55 @@ def add_arguments(self, parser) -> None:
             default=False,
             help="Enable CPU offloading for diffusion models to reduce GPU memory usage.",
         )
+        add_argument(
+            g,
+            flag_name="--task-type",
+            env_var="DYN_OMNI_TASK_TYPE",
+            default=None,
+            help=(
+                "Model-defined startup task partition. MiniMax-H3 accepts "
+                "'fl2va' or 'ref2va'; omit it to load both partitions."
+            ),
+        )
+        add_argument(
+            g,
+            flag_name="--diffusion-attention-backend",
+            env_var="DYN_OMNI_DIFFUSION_ATTENTION_BACKEND",
+            default=None,
+            help="Default vLLM-Omni diffusion attention backend.",
+        )
+        add_argument(
+            g,
+            flag_name="--diffusion-attention-config",
+            env_var="DYN_OMNI_DIFFUSION_ATTENTION_CONFIG",
+            default=None,
+            help="vLLM-Omni diffusion attention configuration as JSON.",
+        )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--enable-distributed-layerwise-offload",
+            env_var="DYN_OMNI_ENABLE_DISTRIBUTED_LAYERWISE_OFFLOAD",
+            default=False,
+            help="Enable distributed layerwise DiT offload.",
+        )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--dlo-use-allgather",
+            env_var="DYN_OMNI_DLO_USE_ALLGATHER",
+            default=True,
+            help="Reconstruct distributed offload shards with AllGather.",
+        )
+        add_argument(
+            g,
+            flag_name="--dlo-resident-layers",
+            env_var="DYN_OMNI_DLO_RESIDENT_LAYERS",
+            default=0,
+            arg_type=int,
+            help=(
+                "Number of leading main-DiT blocks kept resident with DLO; "
+                "positive values require --no-dlo-use-allgather."
+            ),
+        )
         add_negatable_bool_argument(
             g,
             flag_name="--enforce-eager",
@@ -436,6 +491,12 @@ def validate(self) -> None:
             raise ValueError("--text-encoder-tp-size must be > 0")
         if not (0 < self.diffusion.boundary_ratio <= 1):
             raise ValueError("--boundary-ratio must be in (0, 1]")
+        if self.diffusion.dlo_resident_layers < 0:
+            raise ValueError("--dlo-resident-layers must be >= 0")
+        if self.diffusion.dlo_resident_layers > 0 and self.diffusion.dlo_use_allgather:
+            raise ValueError(
+                "--dlo-resident-layers > 0 requires --no-dlo-use-allgather"
+            )
         if self.stage_configs_path is None:
             if self.stage_id is not None:
                 raise ValueError("--stage-id requires --stage-configs-path")