Add H3 support into Dreamverse

apps/dreamverse/README.md

@@ -97,13 +97,33 @@ dreamverse-server --port 8009
 dreamverse-mock-server --port 8009
 ```
 
+### Run Dreamverse with FastH3
+
+Select the VSA data-free FastH3 Preview profile when you start the backend:
+
+```bash
+DREAMVERSE_MODEL_ID=fast-h3 dreamverse-server --port 8009
+```
+
+The `fast-h3` profile uses four visible GPUs by default. It loads the `MiniMaxAI/MiniMax-H3` base checkpoint and the
+`vsa-datafree/adapter_model.safetensors` adapter from
+`FastVideo/FastVideo-FastH3-4-step-Preview-v1-LoRA`. Each request generates a 124-frame, 768×1344 video with
+synchronized audio and five sigma-grid points. Dreamverse uses the last frame of each segment as first-frame
+conditioning for the following segment.
+
+Set `CUDA_VISIBLE_DEVICES` when you need to choose the four physical GPUs:
+
+```bash
+CUDA_VISIBLE_DEVICES=0,1,2,3 DREAMVERSE_MODEL_ID=fast-h3 dreamverse-server --port 8009
+```
+
 > **Expect a slow first boot.** With `torch.compile` and startup warmup enabled
 > (the default), the backend compiles the segment 1 and segment 2 inference
 > paths before it reports ready — this can take **tens of minutes on a cold
 > cache**, regardless of how you deploy (local, server, Docker, or Modal).
 > `/healthz` responds as soon as the process is up; `/readyz` stays `503` until
-> warmup finishes. For a faster, uncompiled startup while testing, set
-> `FASTVIDEO_ENABLE_STARTUP_WARMUP=0` before starting the backend.
+> warmup finishes. To defer compilation until the first generated request while
+> testing, set `FASTVIDEO_ENABLE_STARTUP_WARMUP=0` before starting the backend.
 
 ## Frontend Setup
 
@@ -219,6 +239,7 @@ selection, and mock-server behavior:
 pytest apps/dreamverse/dreamverse/tests/test_config.py \
   apps/dreamverse/dreamverse/tests/test_entrypoints.py \
   apps/dreamverse/dreamverse/tests/test_gpu_pool.py \
+  apps/dreamverse/dreamverse/tests/test_minimax_h3_generation.py \
   apps/dreamverse/dreamverse/tests/test_mock_server.py -q
 ```
 

apps/dreamverse/arch.md

@@ -139,7 +139,18 @@ session.
 - startup warmup
 - user join/leave commands
 - `USER_STEP` execution for each segment
-- continuation state between segments
+- generation-command routing and stream-result delivery
+
+Model generation has a separate ownership boundary inside each GPU process:
+
+- `apps/dreamverse/dreamverse/generation_worker.py` selects the backend that the active model profile declares and owns
+  the backend lifecycle.
+- `apps/dreamverse/dreamverse/ltx2_generation.py` owns LTX-2 generator configuration, video and audio continuation, and
+  runtime LoRA application.
+- `apps/dreamverse/dreamverse/minimax_h3_generation.py` owns the VSA data-free FastH3 adapter, FastH3 generator and
+  request configuration, and last-frame continuation through MiniMax H3 first-frame conditioning.
+- `apps/dreamverse/dreamverse/generation_contracts.py` defines the decoded media and stream-trimming result that both
+  model backends return to `apps/dreamverse/dreamverse/gpu_pool.py`.
 
 `apps/dreamverse/dreamverse/prompt_enhancer.py` manages:
 

apps/dreamverse/dreamverse/benchmarks/benchmark_pipeline.py

@@ -1,6 +1,6 @@
 """Benchmark the LTX-2 generation pipeline driven by the dreamverse Python SDK path.
 
-Mirrors how ``apps/dreamverse/dreamverse/video_generation.py`` constructs
+Mirrors how ``apps/dreamverse/dreamverse/ltx2_generation.py`` constructs
 ``GeneratorConfig`` and calls ``VideoGenerator.generate()``, then
 captures per-stage timings via the ``FASTVIDEO_STAGE_LOGGING=1`` log
 hooks (same mechanism as ``FastVideo-internal/examples/inference/basic/

apps/dreamverse/dreamverse/config.py

@@ -1,5 +1,6 @@
 import os
 from pathlib import Path
+from typing import cast
 
 _REPO_ROOT = Path(__file__).resolve().parents[1]
 _SERVER_ROOT = Path(__file__).resolve().parent
@@ -55,16 +56,34 @@ def _resolve_frontend_static_dir_candidates() -> tuple[str, ...]:
 MODEL_REGISTRY = {
     "fast-ltx2": {
         "name": "FastLTX2",
+        "generation_backend": "ltx2",
+        "default_sp_size": 1,
         "model_path": "FastVideo/LTX2-Distilled-Diffusers",
         "config_model_path": "FastVideo/LTX2-Distilled-Diffusers",
         "lora_repo": "FastVideo/LTX2-OmniNFT-LoRA",
     },
     "fast-ltx23": {
         "name": "FastLTX23",
+        "generation_backend": "ltx2",
+        "default_sp_size": 1,
         "model_path": "FastVideo/LTX-2.3-Distilled-Diffusers",
         "config_model_path": "FastVideo/LTX-2.3-Distilled-Diffusers",
         "lora_repo": "FastVideo/LTX-2.3-OmniNFT-LoRA",
     },
+    "fast-h3": {
+        "name": "FastH3",
+        "generation_backend": "minimax_h3",
+        "default_sp_size": 4,
+        "model_path": "MiniMaxAI/MiniMax-H3",
+        "adapter_repo": "FastVideo/FastVideo-FastH3-4-step-Preview-v1-LoRA",
+        "adapter_filename": "vsa-datafree/adapter_model.safetensors",
+        "attention_backend": "VIDEO_SPARSE_ATTN_H3",
+        "height": 768,
+        "width": 1344,
+        "num_frames": 124,
+        "num_inference_steps": 5,
+        "seed": 1000,
+    },
 }
 
 DEFAULT_MODEL_ID = "fast-ltx2"
@@ -171,7 +190,7 @@ def _optional_env(*names: str) -> str | None:
 DEVTOOLS_ENABLED = _env_bool("FASTVIDEO_ENABLE_DEVTOOLS", False)
 PROMPT_SAFETY_ENABLED = _env_bool("FASTVIDEO_ENABLE_PROMPT_SAFETY", False)
 DREAMVERSE_MAX_AUTOTUNE = _env_bool("DREAMVERSE_MAX_AUTOTUNE", True)
-DREAMVERSE_SP_SIZE = max(1, _env_int("DREAMVERSE_SP_SIZE", 1))
+DREAMVERSE_SP_SIZE = max(1, _env_int("DREAMVERSE_SP_SIZE", cast(int, MODEL_CONFIG["default_sp_size"])))
 
 DREAMVERSE_MODEL_PATH = (os.getenv("DREAMVERSE_MODEL_PATH", "").strip() or None)
 if DREAMVERSE_MODEL_PATH:
@@ -213,7 +232,7 @@ def _resolve_lora_spec(spec: str) -> str | None:
     if not spec:
         return None
     if spec.lower() == "omninft":
-        return MODEL_CONFIG.get("lora_repo")
+        return cast(str | None, MODEL_CONFIG.get("lora_repo"))
     if spec.lower() in AVAILABLE_LORAS:
         return AVAILABLE_LORAS[spec.lower()]["repo"]
     return spec

apps/dreamverse/dreamverse/generation_contracts.py

@@ -0,0 +1,46 @@
+"""Shared contract between DreamVerse generation backends and GPU workers."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+
+@dataclass
+class StepResult:
+    """Decoded media and stream-trimming metadata for one DreamVerse segment."""
+
+    frames: list
+    audio: Any
+    audio_sample_rate: int | None
+    timings: dict[str, float]
+    head_trim_frames: int
+    head_trim_audio_frames: int
+
+
+class GenerationBackend(Protocol):
+    """Model-owned generation operations used by one GPU worker process."""
+
+    def initialize(self, model_config: dict | None = None) -> None:
+        ...
+
+    def shutdown(self) -> None:
+        ...
+
+    def clear_conditioning(self) -> None:
+        ...
+
+    def generate_step(
+        self,
+        prompt: str,
+        segment_idx: int,
+        image_path: str | None,
+        reset_conditioning: bool,
+    ) -> StepResult:
+        ...
+
+    def warmup(self, prompt: str) -> dict[str, float]:
+        ...
+
+    def apply_lora_stack(self, stack: list[tuple[str, float]]) -> tuple[str | None, str | None]:
+        ...

apps/dreamverse/dreamverse/generation_worker.py

@@ -0,0 +1,96 @@
+"""Select and own one model-specific generation backend per GPU process."""
+
+from __future__ import annotations
+
+from dreamverse.config import MODEL_CONFIG
+from dreamverse.generation_contracts import GenerationBackend, StepResult
+
+
+def _create_generation_backend(backend_name: str, gpu_id: int) -> GenerationBackend:
+    """Construct the backend that owns the selected model family's behavior."""
+    if backend_name == "ltx2":
+        from dreamverse.ltx2_generation import LTX2GenerationBackend
+
+        return LTX2GenerationBackend(gpu_id)
+    if backend_name == "minimax_h3":
+        from dreamverse.minimax_h3_generation import MiniMaxH3GenerationBackend
+
+        return MiniMaxH3GenerationBackend(gpu_id)
+    raise ValueError(f"Unsupported DreamVerse generation backend: {backend_name!r}")
+
+
+class VideoGenerationWorker:
+    """Delegate GPU lifecycle and generation calls to the active model backend."""
+
+    def __init__(self, gpu_id: int):
+        self.gpu_id = gpu_id
+        self.model_config: dict = dict(MODEL_CONFIG)
+        self.backend_name: str | None = None
+        self.backend: GenerationBackend | None = None
+
+    def initialize(self, model_config: dict | None = None) -> None:
+        """Load the requested model through its generation backend.
+
+        Model selection belongs here so the GPU process and streaming layers
+        use one stable media contract without importing model-specific code.
+        """
+        requested_model_config = dict(model_config) if model_config is not None else dict(self.model_config)
+        backend_name = requested_model_config.get("generation_backend")
+        if not isinstance(backend_name, str) or not backend_name:
+            raise ValueError("DreamVerse model configuration requires `generation_backend`.")
+
+        candidate_backend = self.backend
+        if candidate_backend is None or self.backend_name != backend_name:
+            if candidate_backend is not None:
+                candidate_backend.shutdown()
+            candidate_backend = _create_generation_backend(backend_name, self.gpu_id)
+
+        try:
+            candidate_backend.initialize(requested_model_config)
+        except Exception:
+            try:
+                candidate_backend.shutdown()
+            except Exception as shutdown_error:
+                print(f"[GPU {self.gpu_id}] Backend cleanup after initialization failure: {shutdown_error}")
+            self.backend = None
+            self.backend_name = None
+            raise
+
+        self.model_config = requested_model_config
+        self.backend = candidate_backend
+        self.backend_name = backend_name
+
+    def _require_backend(self) -> GenerationBackend:
+        """Return the initialized backend or fail before processing a command."""
+        if self.backend is None:
+            raise RuntimeError("Generation backend is not initialized.")
+        return self.backend
+
+    def shutdown(self) -> None:
+        """Release model resources owned by the selected backend."""
+        if self.backend is not None:
+            self.backend.shutdown()
+
+    def clear_conditioning(self) -> None:
+        self._require_backend().clear_conditioning()
+
+    def generate_step(
+        self,
+        prompt: str,
+        segment_idx: int,
+        image_path: str | None,
+        reset_conditioning: bool,
+    ) -> StepResult:
+        """Generate one segment through the selected model backend."""
+        return self._require_backend().generate_step(
+            prompt,
+            segment_idx,
+            image_path,
+            reset_conditioning,
+        )
+
+    def warmup(self, prompt: str) -> dict[str, float]:
+        return self._require_backend().warmup(prompt)
+
+    def apply_lora_stack(self, stack: list[tuple[str, float]]) -> tuple[str | None, str | None]:
+        return self._require_backend().apply_lora_stack(stack)

apps/dreamverse/dreamverse/gpu_pool.py

@@ -12,7 +12,7 @@
 from multiprocessing import Process, Queue
 
 from dreamverse.config import (
-    DEFAULT_MODEL_ID,
+    ACTIVE_MODEL_ID,
     DREAMVERSE_SP_SIZE,
     MODEL_REGISTRY,
     STARTUP_WARMUP_ENABLED,
@@ -54,7 +54,7 @@
 def _parse_requested_gpu_limit() -> int | None:
     raw_value = os.getenv("FASTVIDEO_GPU_COUNT", "").strip().lower()
     if not raw_value:
-        return 1
+        return DREAMVERSE_SP_SIZE
     if raw_value == "all":
         return None
     try:
@@ -164,12 +164,12 @@ def gpu_worker_process(
     os.environ["CUDA_VISIBLE_DEVICES"] = cuda_device
     os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "FLASH_ATTN"
 
-    from dreamverse.video_generation import VideoGenerationWorker
+    from dreamverse.generation_worker import VideoGenerationWorker
 
     worker = VideoGenerationWorker(gpu_id)
 
     def event_loop(first_cmd: Command = None):
-        """Blocking event loop for LTX2; dispatches user commands."""
+        """Block on generation commands after the model is initialized."""
         print(f"[GPU {gpu_id}] Entering event loop")
 
         def handle_command(cmd: Command):
@@ -435,7 +435,7 @@ def __init__(self, gpu_id: int, cuda_device: str):
         self._response_reader_task: asyncio.Task | None = None
         self._active: bool = False
         self._reader_lock: asyncio.Lock | None = None
-        self.current_model_id: str = DEFAULT_MODEL_ID
+        self.current_model_id: str | None = ACTIVE_MODEL_ID
         self.shared_stream_buffer = None
         self.shared_stream_buffer_size = SHARED_STREAM_BUFFER_BYTES
 
@@ -690,7 +690,7 @@ def unregister_stream_queue(self, user_id: str) -> None:
     async def join_user(self, user_id: str, model_id: str = None) -> JoinAck:
         """Add a user to this GPU."""
         if model_id is None:
-            model_id = DEFAULT_MODEL_ID
+            model_id = ACTIVE_MODEL_ID
 
         # Reload model if a different one is requested
         if model_id != self.current_model_id and model_id in MODEL_REGISTRY:
@@ -705,16 +705,23 @@ async def join_user(self, user_id: str, model_id: str = None) -> JoinAck:
             self.connected_users.clear()
 
             model_config = MODEL_REGISTRY[model_id]
-            reload_response = await self._send_command(Command(CommandType.RELOAD_MODEL,
-                                                               payload=ReloadModelPayload(model_config=model_config),
-                                                               user_id="__reload__"),
-                                                       timeout=600.0)
+            try:
+                reload_response = await self._send_command(Command(
+                    CommandType.RELOAD_MODEL,
+                    payload=ReloadModelPayload(model_config=model_config),
+                    user_id="__reload__"),
+                                                           timeout=600.0)
+            except Exception:
+                self.current_model_id = None
+                raise
             match reload_response:
                 case ReloadAck():
                     pass
                 case WorkerError(message=msg):
+                    self.current_model_id = None
                     raise RuntimeError(f"Model reload failed: {msg}")
                 case _:
+                    self.current_model_id = None
                     raise RuntimeError(f"Unexpected reload response: "
                                        f"{type(reload_response).__name__}")