[perf]: Release H3 text encoder before DiT/VAE

GB10 unified memory disables host offload, so MiniMax H3 CUDA used
to load Qwen3-VL together with the DiT and VAEs and get killed by
earlyoom. Encode first, drop the encoder, then load denoise weights.
Input-prep geometry comes from the VAE arch configs until those
modules exist.

A later generate() on the same worker still needs a new process;
prompt-cache reload is not in this change.

Author: aryan5v

docs/getting_started/installation/spark_performance.md

@@ -161,6 +161,11 @@ is power-cycled. To avoid it:
   on: "CPU" offload uses the same unified RAM. Multi-GPU FSDP sharding remains
   available because it partitions weights without parking them in a separate
   host pool.
+- **MiniMax H3 / FastH3** still needs sequential loading on one GB10. The Qwen3-VL
+  conditioner is tens of gigabytes of BF16. If the DiT and VAEs load while that
+  encoder is still resident, the process is a typical `earlyoom` kill (Python is
+  preferred). The CUDA pipeline now encodes first, releases the encoder, then
+  loads DiT and VAEs. See [Offloading](../../inference/offloading.md).
 
 ## Gotchas specific to the GB10
 
@@ -177,6 +182,10 @@ A few things that surprise people on this box (beyond the memory notes above):
 - **Cosmos-2.5** uses a Qwen2.5-VL text encoder; make sure you're on a FastVideo
   build recent enough to include its `transformers`-compatibility handling before
   running it.
+- **MiniMax H3 worker init can look healthy and still die on the first generate**
+  if you are on a build that loads encoder, VAE, and DiT together. Confirm the log
+  contains `Released MiniMax-H3 text encoder after conditioning` before
+  `Loading MiniMax-H3 denoise modules`.
 
 ## Reproduce these numbers
 

docs/inference/offloading.md

@@ -21,6 +21,15 @@ pool there, so offload adds transfers and duplicate residency instead of freeing
 memory. CUDA FSDP sharding remains enabled when requested; MPS continues to
 disable FSDP. `pin_cpu_memory` is not an offload mode and is left unchanged.
 
+MiniMax H3 CUDA inference uses a second lever that does not copy weights to a
+host pool. The pipeline loads the Qwen3-VL text encoder, runs conditioning, then
+releases that encoder before it loads the DiT and video/audio VAEs. The MLX FastH3
+runtime uses the same phase order. Input-preparation geometry (spatial ratio,
+latent channels, audio sample rate) comes from the VAE arch configs until those
+weights load. A later `generate()` on the same worker currently re-enters
+conditioning after the encoder has been released; start a new generator for a
+new prompt until prompt-cache reload exists.
+
 ## Behavior Explanation
 
 !!! note

fastvideo/pipelines/basic/minimax_h3/minimax_h3_pipeline.py

@@ -3,8 +3,17 @@
 
 from __future__ import annotations
 
+import gc
+from dataclasses import dataclass
+from typing import Any
+
+import torch
+
+from fastvideo.configs.models.vaes.minimax_h3_audio import MiniMaxH3AudioVAEArchConfig
+from fastvideo.configs.models.vaes.minimax_h3_video import MiniMaxH3VideoVAEArchConfig
 from fastvideo.configs.pipelines.minimax_h3 import MiniMaxH3PipelineConfig
 from fastvideo.fastvideo_args import FastVideoArgs
+from fastvideo.logger import init_logger
 from fastvideo.pipelines.basic.minimax_h3.stages import (
     MiniMaxH3AudioDecodingStage,
     MiniMaxH3ConditioningStage,
@@ -15,6 +24,37 @@
 )
 from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
 from fastvideo.pipelines.lora_pipeline import LoRAPipeline
+from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
+
+logger = init_logger(__name__)
+
+# Same split as the MLX runtime: condition, release the ~66 GB Qwen3-VL stack,
+# then load DiT + VAEs. Keeping them resident together OOMs unified-memory
+# boxes (GB10 / Spark) even though host offload is correctly disabled there.
+_DENOISE_MODULE_NAMES = ("vae", "audio_vae", "transformer")
+
+
+@dataclass(frozen=True)
+class _H3VideoGeometry:
+    spatial_compression_ratio: int
+    latent_channels: int
+
+
+@dataclass(frozen=True)
+class _H3AudioGeometry:
+    sampling_rate: int
+
+
+def _default_video_geometry() -> _H3VideoGeometry:
+    arch = MiniMaxH3VideoVAEArchConfig()
+    return _H3VideoGeometry(
+        spatial_compression_ratio=int(arch.spatial_compression_ratio),
+        latent_channels=int(arch.latent_channels),
+    )
+
+
+def _default_audio_geometry() -> _H3AudioGeometry:
+    return _H3AudioGeometry(sampling_rate=int(MiniMaxH3AudioVAEArchConfig().sampling_rate))
 
 
 class MiniMaxH3BasePipeline(LoRAPipeline, ComposedPipelineBase):
@@ -53,6 +93,11 @@ class MiniMaxH3BasePipeline(LoRAPipeline, ComposedPipelineBase):
         "audio_scheduler",
     ]
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self._ref2va = False
+        self._denoise_stages_ready = False
+        super().__init__(*args, **kwargs)
+
     @classmethod
     def get_hf_download_component_dirs(cls) -> tuple[str, ...]:
         return tuple(sorted(cls._extra_config_module_map.get(name, name) for name in cls._required_config_modules))
@@ -67,18 +112,70 @@ def initialize_pipeline(self, fastvideo_args: FastVideoArgs) -> None:
             if shift is None or float(shift) != expected_shift:
                 raise ValueError(f"MiniMax-H3 {modality} scheduler must expose shift={expected_shift:g}, got {shift}.")
 
-    def _add_stages(self, *, ref2va: bool) -> None:
-        transformer = self.get_module("transformer")
-        vae = self.get_module("vae")
-        audio_vae = self.get_module("audio_vae")
-        scheduler = self.get_module("scheduler")
-        audio_scheduler = self.get_module("audio_scheduler")
+    def _defer_denoise_modules(self, fastvideo_args: FastVideoArgs) -> bool:
+        return bool(fastvideo_args.inference_mode) and not bool(getattr(fastvideo_args, "training_mode", False))
+
+    def _denoise_modules_loaded(self) -> bool:
+        return all(self.get_module(name) is not None for name in _DENOISE_MODULE_NAMES)
 
+    def load_modules(self,
+                     fastvideo_args: FastVideoArgs,
+                     loaded_modules: dict[str, torch.nn.Module] | None = None) -> dict[str, Any]:
+        """Load the Qwen3-VL conditioner first; defer DiT and VAEs until after encode."""
+        if not self._defer_denoise_modules(fastvideo_args):
+            return super().load_modules(fastvideo_args, loaded_modules)
+        if loaded_modules is not None and all(name in loaded_modules for name in _DENOISE_MODULE_NAMES):
+            return super().load_modules(fastvideo_args, loaded_modules)
+
+        saved = list(self.required_config_modules)
+        self._required_config_modules = [name for name in saved if name not in _DENOISE_MODULE_NAMES]
+        try:
+            logger.info("Loading MiniMax-H3 condition modules first: %s", self._required_config_modules)
+            return super().load_modules(fastvideo_args, loaded_modules)
+        finally:
+            self._required_config_modules = saved
+
+    def _load_denoise_modules(self, fastvideo_args: FastVideoArgs) -> None:
+        if self._denoise_modules_loaded():
+            return
+        saved = list(self.required_config_modules)
+        self._required_config_modules = [name for name in saved if name != "text_encoder"]
+        try:
+            logger.info("Loading MiniMax-H3 denoise modules after releasing the text encoder: %s",
+                        [name for name in self._required_config_modules if name in _DENOISE_MODULE_NAMES])
+            loaded = super().load_modules(fastvideo_args, loaded_modules=self.modules)
+            for name, module in loaded.items():
+                self.add_module(name, module)
+        finally:
+            self._required_config_modules = saved
+
+    def _release_text_encoder(self) -> None:
+        stage = self._stage_name_mapping.get("conditioning_stage")
+        if stage is not None:
+            stage.conditioner = None
+        encoder = self.modules.pop("text_encoder", None)
+        if encoder is None:
+            return
+        logger.info("Released MiniMax-H3 text encoder after conditioning")
+        del encoder
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+
+    def _input_vae(self) -> Any:
+        return self.get_module("vae") or _default_video_geometry()
+
+    def _input_audio_vae(self, *, ref2va: bool) -> Any | None:
+        if not ref2va:
+            return None
+        return self.get_module("audio_vae") or _default_audio_geometry()
+
+    def _add_condition_stages(self, *, ref2va: bool) -> None:
         self.add_stage(
             "input_preparation_stage",
             MiniMaxH3InputPreparationStage(
-                vae=vae,
-                audio_vae=audio_vae if ref2va else None,
+                vae=self._input_vae(),
+                audio_vae=self._input_audio_vae(ref2va=ref2va),
                 ref2va=ref2va,
             ),
         )
@@ -91,6 +188,15 @@ def _add_stages(self, *, ref2va: bool) -> None:
                 ref2va=ref2va,
             ),
         )
+
+    def _add_denoise_stages(self, *, ref2va: bool) -> None:
+        transformer = self.get_module("transformer")
+        vae = self.get_module("vae")
+        audio_vae = self.get_module("audio_vae")
+        scheduler = self.get_module("scheduler")
+        audio_scheduler = self.get_module("audio_scheduler")
+        if transformer is None or vae is None or audio_vae is None:
+            raise RuntimeError("MiniMax-H3 denoise stages require transformer, vae, and audio_vae to be loaded.")
         self.add_stage(
             "latent_preparation_stage",
             MiniMaxH3LatentPreparationStage(
@@ -111,6 +217,35 @@ def _add_stages(self, *, ref2va: bool) -> None:
         )
         self.add_stage("video_decoding_stage", MiniMaxH3VideoDecodingStage(vae=vae, transformer=transformer))
         self.add_stage("audio_decoding_stage", MiniMaxH3AudioDecodingStage(audio_vae=audio_vae))
+        self._denoise_stages_ready = True
+
+    def _add_stages(self, *, ref2va: bool) -> None:
+        self._ref2va = ref2va
+        self._add_condition_stages(ref2va=ref2va)
+        if self._denoise_modules_loaded():
+            self._add_denoise_stages(ref2va=ref2va)
+
+    def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
+        if not self.post_init_called:
+            self.post_init()
+
+        if self._denoise_stages_ready:
+            return super().forward(batch, fastvideo_args)
+
+        logger.info("Running MiniMax-H3 condition stages before loading DiT/VAE weights")
+        for stage in self.stages:
+            batch = stage(batch, fastvideo_args)
+        self._release_text_encoder()
+        self._load_denoise_modules(fastvideo_args)
+        self._add_denoise_stages(ref2va=self._ref2va)
+        for name in (
+                "latent_preparation_stage",
+                "denoising_stage",
+                "video_decoding_stage",
+                "audio_decoding_stage",
+        ):
+            batch = self._stage_name_mapping[name](batch, fastvideo_args)
+        return batch
 
 
 class MiniMaxH3Pipeline(MiniMaxH3BasePipeline):

fastvideo/tests/stages/test_minimax_h3_sequential_start.py

@@ -0,0 +1,130 @@
+# SPDX-License-Identifier: Apache-2.0
+"""CPU contracts for MiniMax-H3 Mac-style sequential module loading."""
+from __future__ import annotations
+
+from contextlib import nullcontext
+from types import SimpleNamespace
+
+import torch
+
+import fastvideo.pipelines.composed_pipeline_base as composed_pipeline_base
+from fastvideo.fastvideo_args import FastVideoArgs
+from fastvideo.pipelines.basic.minimax_h3.minimax_h3_pipeline import (
+    MiniMaxH3Pipeline,
+    _DENOISE_MODULE_NAMES,
+)
+from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
+from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
+
+
+class _Profiler:
+
+    def region(self, name):
+        del name
+        return nullcontext()
+
+
+def _stub_module(name: str) -> SimpleNamespace:
+    if name in {"scheduler"}:
+        return SimpleNamespace(shift=12.0, name=name)
+    if name in {"audio_scheduler"}:
+        return SimpleNamespace(shift=3.0, name=name)
+    if name == "transformer":
+        # LoRAPipeline reads exclude_lora_layers off the DiT arch config.
+        return SimpleNamespace(
+            name=name,
+            config=SimpleNamespace(arch_config=SimpleNamespace(exclude_lora_layers=[])),
+        )
+    return SimpleNamespace(name=name)
+
+
+def _patch_pipeline_construction(monkeypatch, events: list) -> None:
+    monkeypatch.setattr(
+        composed_pipeline_base,
+        "maybe_init_distributed_environment_and_model_parallel",
+        lambda *args, **kwargs: events.append(("distributed", None)),
+    )
+    monkeypatch.setattr(composed_pipeline_base, "get_local_torch_device", lambda: torch.device("cpu"))
+    monkeypatch.setattr(composed_pipeline_base, "get_world_group", lambda: SimpleNamespace(local_rank=0))
+    monkeypatch.setattr(composed_pipeline_base, "get_or_create_profiler", lambda trace_dir: _Profiler())
+    monkeypatch.setattr(composed_pipeline_base, "warmup_sequence_parallel_communication", lambda: None)
+    monkeypatch.setattr("fastvideo.platforms.current_platform.has_unified_memory", lambda device_id: False)
+    monkeypatch.setattr("fastvideo.platforms.current_platform.is_mps", lambda: False)
+
+
+def test_inference_defers_dit_and_vae_until_after_conditioning(monkeypatch) -> None:
+    events: list = []
+    _patch_pipeline_construction(monkeypatch, events)
+    loads: list[list[str]] = []
+
+    def fake_load(self, fastvideo_args, loaded_modules=None):
+        del fastvideo_args
+        requested = list(self.required_config_modules)
+        loads.append(requested)
+        modules = dict(loaded_modules or {})
+        for name in requested:
+            modules.setdefault(name, _stub_module(name))
+        return modules
+
+    monkeypatch.setattr(ComposedPipelineBase, "load_modules", fake_load)
+
+    args = FastVideoArgs(model_path="unused/for-this-test", enable_stage_verification=False)
+    pipeline = MiniMaxH3Pipeline("unused/for-this-test", args)
+    pipeline.post_init()
+
+    assert loads, "condition modules should load during construction"
+    assert "text_encoder" in loads[0]
+    assert all(name not in loads[0] for name in _DENOISE_MODULE_NAMES)
+    assert pipeline.get_module("text_encoder") is not None
+    assert pipeline.get_module("transformer") is None
+    assert list(pipeline._stage_name_mapping) == ["input_preparation_stage", "conditioning_stage"]
+
+    condition_stage = pipeline._stage_name_mapping["conditioning_stage"]
+    passthrough = lambda batch, _args: batch
+    monkeypatch.setattr(pipeline._stage_name_mapping["input_preparation_stage"], "forward", passthrough)
+    monkeypatch.setattr(condition_stage, "forward", passthrough)
+
+    original_add_denoise = pipeline._add_denoise_stages
+
+    def fake_add_denoise(*, ref2va: bool) -> None:
+        original_add_denoise(ref2va=ref2va)
+        for name in (
+                "latent_preparation_stage",
+                "denoising_stage",
+                "video_decoding_stage",
+                "audio_decoding_stage",
+        ):
+            monkeypatch.setattr(pipeline._stage_name_mapping[name], "forward", passthrough)
+
+    monkeypatch.setattr(pipeline, "_add_denoise_stages", fake_add_denoise)
+
+    batch = ForwardBatch(data_type="video", prompt="alpine dancer")
+    out = pipeline.forward(batch, args)
+
+    assert out is batch
+    assert len(loads) == 2
+    assert "transformer" in loads[1]
+    assert "vae" in loads[1]
+    assert "text_encoder" not in loads[1]
+    assert pipeline.get_module("text_encoder") is None
+    assert condition_stage.conditioner is None
+    assert pipeline.get_module("transformer") is not None
+    assert pipeline._denoise_stages_ready is True
+
+
+def test_injected_denoise_weights_skip_the_deferred_split(monkeypatch) -> None:
+    events: list = []
+    _patch_pipeline_construction(monkeypatch, events)
+    loads: list[list[str]] = []
+
+    def fake_load(self, fastvideo_args, loaded_modules=None):
+        del fastvideo_args
+        loads.append(list(self.required_config_modules))
+        return dict(loaded_modules or {})
+
+    monkeypatch.setattr(ComposedPipelineBase, "load_modules", fake_load)
+    injected = {name: _stub_module(name) for name in MiniMaxH3Pipeline._required_config_modules}
+    args = FastVideoArgs(model_path="unused/for-this-test")
+    MiniMaxH3Pipeline("unused/for-this-test", args, loaded_modules=injected)
+
+    assert loads == [list(MiniMaxH3Pipeline._required_config_modules)]