--- title: minimax_h3_pipeline ---

End-to-end MiniMax-H3 joint audio-video generation on Apple Silicon MLX.

Phased runtime that keeps one heavyweight component resident at a time:

  1. Condition — streamed Qwen3-VL text encoding (or a verified prompt embedding cache);
  2. Denoise — pre-quantized H3 DiT (one of int8/int6/int4 resident), dual rectified-flow schedulers (video shift 12 / audio shift 3), served from the persisted AdaLN ladder;
  3. Decode — MLX H3 video VAE and audio VAE, sequentially;
  4. Mux — H.264 24 fps + AAC 32 kHz stereo MP4 via ffmpeg.

MLX-native memory cleanup (mx.clear_cache) runs between every phase. The MLX path itself does not call PyTorch.

fastvideo.mlx_runtime.minimax_h3_pipeline.init_logger

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

fastvideo.mlx_runtime.minimax_h3_pipeline.upsample_frames

upsample_frames(frames: Iterable[np.ndarray], *, width: int, height: int, mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE, sharpen: float = 0.0) -> list[np.ndarray]

Resample every decoded frame to the target pixel size.

Parameters

frames: Iterable[np.ndarray]

Decoded HxWx3 uint8 RGB frames.

width: int

Target width in pixels.

height: int

Target height in pixels.

mode: str

Interpolation kernel, one of :data:PIXEL_UPSAMPLE_MODES.

sharpen: float

Unsharp strength applied after each resize.

Returns

list[np.ndarray]

list[np.ndarray]: New frames at the target size, in input order.

fastvideo.mlx_runtime.minimax_h3_pipeline.set_quant_gemm_mode

set_quant_gemm_mode(mode: str | None) -> str | None

Select the H3 affine-quantized GEMM path process-wide; returns the previous override.

auto keeps the measured default (FASTVIDEO_MLX_DQ_GEMM decides), dequant always uses dequantize + dense GEMM for wide inputs, and qmm always uses mx.quantized_matmul. None clears the override so FASTVIDEO_MLX_QUANT_GEMM applies again.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3SchedulerState

MiniMaxH3SchedulerState(shift: float, sigmas: np.ndarray, timesteps: np.ndarray)

One rectified-flow scheduler (use two: video shift 12, audio shift 3).

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3SchedulerState.create

create(shift: float, num_denoise_steps: int) -> MiniMaxH3SchedulerState

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3SchedulerState.from_dmd_steps

from_dmd_steps(shift: float, dmd_steps: list[int]) -> MiniMaxH3SchedulerState

Shift explicit DMD rungs the way the CUDA denoiser does.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3SchedulerState.from_pdd_node_sigmas

from_pdd_node_sigmas(shift: float, node_sigmas: np.ndarray) -> MiniMaxH3SchedulerState

Carry a PDD plan's shifted node sigmas as the sampling ladder.

node_sigmas are already shifted (one noise level per fused-block node, descending to 0), so an ordinary Euler step between consecutive nodes applies exactly the block's total integration weight.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3SchedulerState.step

step(model_output, step_index: int, sample)

Data-ward Euler: x0 = x_t + sigma*v; blend toward x0 by sigma ratio.

fp32 blend for fp16/bf16 samples, matching the reference.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3SchedulerState.scale_noise

scale_noise(sample, timestep: float, noise)

Conditioning noise-aug: tsample + (1-t)noise (t=0.999 for keyframes).

fastvideo.mlx_runtime.minimax_h3_pipeline.adaln_timestep_union

adaln_timestep_union(video: MiniMaxH3SchedulerState, audio: MiniMaxH3SchedulerState) -> np.ndarray

Sorted unique AdaLN timesteps for one video/audio pair, plus clean time 1.

fastvideo.mlx_runtime.minimax_h3_pipeline.align_num_frames

align_num_frames(num_frames: int) -> int

fastvideo.mlx_runtime.minimax_h3_pipeline.audio_latent_num_frames

audio_latent_num_frames(num_frames: int) -> int

fastvideo.mlx_runtime.minimax_h3_pipeline.build_packed_layout

build_packed_layout(num_text_tokens: int, num_latent_frames: int, latent_height: int, latent_width: int, num_audio_latents: int, patch_size: tuple[int, int, int] = (1, 2, 2), keyframe_anchors: tuple[str, ...] = (), text_token_tags: np.ndarray | None = None, video_temporal_scale: float = 1.0) -> MiniMaxH3PackedLayout

Build the [text | condition | audio | video] layout (T2VA / FL2VA).

fastvideo.mlx_runtime.minimax_h3_pipeline.build_ref2va_packed_layout

build_ref2va_packed_layout(text_token_tags: np.ndarray, references: list[MiniMaxH3PreparedReference], num_latent_frames: int, latent_height: int, latent_width: int, num_audio_latents: int, patch_size: tuple[int, int, int] = (1, 2, 2)) -> MiniMaxH3PackedLayout

Build the [text | ordered references | target audio | target video] layout.

Mirrors fastvideo.pipelines.basic.minimax_h3.packing.build_ref2va_packed_sequence so a reference-conditioned request packs the same rows, positions and regions on both runtimes. References keep their declared order and each one advances the shared rotary clock by what it occupies, which is what makes the target rows continue after them rather than overlap.

fastvideo.mlx_runtime.minimax_h3_pipeline.build_row_timesteps

build_row_timesteps(layout: MiniMaxH3PackedLayout, video_timestep: float, audio_timestep: float, condition_video_timestep: float = 1.0, condition_audio_timestep: float = 1.0) -> tuple[np.ndarray, np.ndarray]

Per-row timesteps -> (unique timesteps sorted ascending, per-row indices).

Matches torch.unique(..., sorted=True, return_inverse=True).

fastvideo.mlx_runtime.minimax_h3_pipeline.dense_only_vsa_error

dense_only_vsa_error(checkpoint_dir: str | Any) -> DenseOnlyVSACheckpointError

fastvideo.mlx_runtime.minimax_h3_pipeline.load_fasth3_inference_contract

load_fasth3_inference_contract(model_root: str | Path | None) -> dict[str, Any] | None

Load a shift-declaring FastH3 contract, DMD or Ref2VA PDD.

Four-step exports ship a sidecar without scheduler shifts and keep the uniform ladder. A contract that declares both shifts selects an explicit ladder: a FastH3 Ref2VA PDD export (pdd_steps / pdd_step_indices) is tagged kind="pdd" and a FastH3 8-Step V2 DMD export (which carries dmd_denoising_steps) is tagged kind="dmd". A shifted sidecar that cannot be parsed fails instead of silently baking the four-step grid.

fastvideo.mlx_runtime.minimax_h3_pipeline.load_mlx_h3_checkpoint

load_mlx_h3_checkpoint(checkpoint_dir: str | Path) -> MLXMiniMaxH3DiT

Rebuild an H3 DiT saved by :func:save_mlx_h3_checkpoint.

Refuses a quantization grid the installed MLX cannot execute, loudly — never silently dequantizes onto a different grid.

fastvideo.mlx_runtime.minimax_h3_pipeline.mlx_h3_checkpoint_vsa_capable

mlx_h3_checkpoint_vsa_capable(checkpoint_dir: str | Path) -> bool

True when a saved MLX H3 checkpoint retained the VSA gate projections.

fastvideo.mlx_runtime.minimax_h3_pipeline.resolve_h3_denoise_schedule

resolve_h3_denoise_schedule(model_root: str | Path | None, num_steps: int, *, cached_timesteps: np.ndarray | None = None) -> ResolvedH3Schedule

Keep the uniform ladder unless the snapshot declares its own shifts.

num_steps is the transformer-forward count. A shift-declaring DMD contract also accepts that count plus one, the CUDA sigma-grid convention, and then runs exactly its trained rung count. A FastH3 Ref2VA PDD contract instead runs its trained fused-block count exactly and the schedulers carry the plan's node sigmas.

fastvideo.mlx_runtime.minimax_h3_pipeline.temporal_position_grid

temporal_position_grid(num_latent_frames: int, origin: float) -> np.ndarray

fastvideo.mlx_runtime.minimax_h3_pipeline.unpatchify_video_tokens

unpatchify_video_tokens(rows: np.ndarray, num_latent_frames: int, latent_height: int, latent_width: int, channels: int, patch_size: tuple[int, int, int]) -> np.ndarray

fastvideo.mlx_runtime.minimax_h3_pipeline.unpack_audio_tokens

unpack_audio_tokens(rows: np.ndarray, num_audio_latents: int) -> np.ndarray

(2*num_audio_latents, feat) channel-major rows -> (2, feat, num_audio_latents).

fastvideo.mlx_runtime.minimax_h3_pipeline.video_latent_num_frames

video_latent_num_frames(num_frames: int) -> int

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3PreparedReference

MiniMaxH3PreparedReference(media_type: Literal['image', 'video', 'audio'], has_audio: bool = False, image: Image.Image | None = None, frames: np.ndarray | None = None, waveform: np.ndarray | None = None, block_timestamps: list[float] = list(), num_latent_frames: int = 1, latent_height: int = 0, latent_width: int = 0, num_audio_latents: int = 0)

Decoded Ref2VA media and the latent geometry resolved by condition encoding.

waveform is (channels, samples) float32, not a tensor: the MLX side never needs a device-resident waveform, the audio VAE converts on encode.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3Reference

MiniMaxH3Reference(source: Any, media_type: Literal['image', 'video', 'audio'] = 'image', soundtrack: Any | None = None, fps: float | None = None, sample_rate: int | None = None)

One ordered Ref2VA input; decoding is deferred until request preparation.

fastvideo.mlx_runtime.minimax_h3_pipeline.prepare_reference

prepare_reference(reference: MiniMaxH3Reference, num_frames: int, target_sample_rate: int, *, max_pixels: int | None = None) -> MiniMaxH3PreparedReference

Decode and normalize one deferred Ref2VA medium; max_pixels caps image/video canvases (down only).

fastvideo.mlx_runtime.minimax_h3_pipeline.validate_references

validate_references(references: list[Any]) -> list[MiniMaxH3Reference]

Validate the ordered Ref2VA media list and its per-modality limits.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3PrefixKVCache

MiniMaxH3PrefixKVCache(num_layers: int, rows: int, num_heads: int, head_dim: int, storage: str = 'bf16', compute_dtype: Any = None, hidden_dtype: Any = None, recorded_step: int = -1, keys: list[Any] = list(), values: list[Any] = list())

Per-block K/V of the frozen rows ([rows, heads, head_dim]) in compact storage.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3PrefixKVCache.reset

reset() -> None

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3PrefixKVCache.store

store(layer: int, key, value) -> list[Any]

Keep the frozen rows of one block's K/V; returns arrays to evaluate.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3PrefixKVCache.load

load(layer: int) -> tuple[Any, Any]

fastvideo.mlx_runtime.minimax_h3_pipeline.check_prefix_kv_budget

check_prefix_kv_budget(estimated_bytes: int, max_gib: float, *, rows: int) -> None

fastvideo.mlx_runtime.minimax_h3_pipeline.estimate_prefix_kv_bytes

estimate_prefix_kv_bytes(num_layers: int, rows: int, num_heads: int, head_dim: int, storage: str = 'bf16', native_itemsize: int = 2) -> int

2 (K and V) * layers * rows * heads * head_dim * itemsize (+ int8 scales).

fastvideo.mlx_runtime.minimax_h3_pipeline.frozen_prefix_rows

frozen_prefix_rows(layout: Any) -> int

Rows before the target audio, after checking the tail is [target audio | target video].

fastvideo.mlx_runtime.minimax_h3_pipeline.validate_prefix_kv_storage

validate_prefix_kv_storage(storage: str) -> str

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3VSAConfig

MiniMaxH3VSAConfig(enabled: bool = False, sparsity: float = 0.9, tile_size: int = 64, prefix_mode: PrefixMode = 'exempt', dense_first_n_steps: int = 0, dense_layers: tuple[int, ...] = (), impl: VSAImpl = 'auto', ref_policy: str | None = None, ref_keep_rate: float | None = None)

Runtime VSA knobs. Defaults preserve dense MLX H3 behavior.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3VSAConfig.effective_ref_keep_rate

effective_ref_keep_rate: float | None

The per-reference-video keep rate, or None when references stay dense.

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3VSAConfig.layer_sparsity

layer_sparsity(layer_idx: int, step_index: int) -> float

fastvideo.mlx_runtime.minimax_h3_pipeline.resolve_vsa_settings

resolve_vsa_settings(contract: dict[str, Any] | None, *, vsa: bool | None = None, sparsity: float | None = None, tile_size: int | None = None, ref_policy: str | None = None, ref_keep_rate: float | None = None, checkpoint_capable: bool = True) -> ResolvedVSASettings

vsa=None means on for a VSA contract (when the checkpoint can) and off otherwise.

fastvideo.mlx_runtime.minimax_h3_pipeline.fingerprint_digest

fingerprint_digest(fingerprint: dict[str, object]) -> str

fastvideo.mlx_runtime.minimax_h3_pipeline.text_encoder_fingerprint

text_encoder_fingerprint(model_root: Path) -> dict[str, object]

Return a cheap identity for the tokenizer and text-encoder files.

fastvideo.mlx_runtime.minimax_h3_pipeline.GenerationResult

GenerationResult(video_path: str | None, frames: np.ndarray | None, waveform: np.ndarray | None, sample_rate: int, timings: dict[str, float] = dict(), peak_memory_gib: dict[str, float] = dict(), vsa: dict[str, Any] = dict(), video_decode_backend: str = 'h3-vae', prefix_kv: dict[str, Any] = dict())
Source
@dataclass
class GenerationResult:
    video_path: str | None
    frames: np.ndarray | None
    waveform: np.ndarray | None
    sample_rate: int
    timings: dict[str, float] = field(default_factory=dict)
    peak_memory_gib: dict[str, float] = field(default_factory=dict)
    vsa: dict[str, Any] = field(default_factory=dict)
    video_decode_backend: str = "h3-vae"
    prefix_kv: dict[str, Any] = field(default_factory=dict)

fastvideo.mlx_runtime.minimax_h3_pipeline.FastTemporalPlan

FastTemporalPlan(target_frames: int, source_frames: int, factor: int, video_temporal_scale: float)

Sparse video geometry for RIFE fast mode with full-duration audio.

Source
@dataclass(frozen=True)
class FastTemporalPlan:
    """Sparse video geometry for RIFE fast mode with full-duration audio."""

    target_frames: int
    source_frames: int
    factor: int
    video_temporal_scale: float

fastvideo.mlx_runtime.minimax_h3_pipeline.plan_fast_temporal

plan_fast_temporal(target_frames: int, factor: int = 2) -> FastTemporalPlan

Choose the smallest H3-valid source sequence that covers the target timeline.

Source
def plan_fast_temporal(target_frames: int, factor: int = 2) -> FastTemporalPlan:
    """Choose the smallest H3-valid source sequence that covers the target timeline."""
    target_frames = align_num_frames(target_frames)
    if factor < 2:
        raise ValueError(f"fast factor must be at least 2, got {factor}.")
    ideal_source_frames = math.ceil((target_frames - 1) / factor) + 1
    source_frames = align_num_frames(ideal_source_frames)
    if source_frames >= target_frames:
        raise ValueError(f"fast factor {factor} does not reduce the H3-aligned target of {target_frames} frames.")

    source_grid = temporal_position_grid(video_latent_num_frames(source_frames), 0.0)
    target_grid = temporal_position_grid(video_latent_num_frames(target_frames), 0.0)
    video_temporal_scale = float(target_grid[-1] / source_grid[-1])
    return FastTemporalPlan(
        target_frames=target_frames,
        source_frames=source_frames,
        factor=factor,
        video_temporal_scale=video_temporal_scale,
    )

fastvideo.mlx_runtime.minimax_h3_pipeline.FastSpatialPlan

FastSpatialPlan(target_height: int, target_width: int, stage1_height: int, stage1_width: int, canvas_height: int, canvas_width: int, scale: int, upsample_mode: str, sharpen: float)

Reduced-canvas geometry for spatial fast mode (RIFE's spatial twin).

Source
@dataclass(frozen=True)
class FastSpatialPlan:
    """Reduced-canvas geometry for spatial fast mode (RIFE's spatial twin)."""

    target_height: int
    target_width: int
    stage1_height: int
    stage1_width: int
    canvas_height: int
    canvas_width: int
    scale: int
    upsample_mode: str
    sharpen: float

fastvideo.mlx_runtime.minimax_h3_pipeline.plan_fast_spatial

plan_fast_spatial(height: int, width: int, *, scale: int = 2, upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE, sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN) -> FastSpatialPlan

Choose the smallest H3-valid canvas that covers target / scale.

H3 geometry rounds up to the 32px model grid and center-crops after decode — the same convention plain 720p generation uses via _model_canvas_size — so no size the full-resolution path accepts is rejected here. The return trip to the target size runs in pixel space after the VAE decode, never on latents; see :mod:fastvideo.mlx_runtime.frame_upsample for why.

Source
def plan_fast_spatial(
    height: int,
    width: int,
    *,
    scale: int = 2,
    upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE,
    sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN,
) -> FastSpatialPlan:
    """Choose the smallest H3-valid canvas that covers ``target / scale``.

    H3 geometry rounds *up* to the 32px model grid and center-crops after
    decode — the same convention plain 720p generation uses via
    ``_model_canvas_size`` — so no size the full-resolution path accepts is
    rejected here. The return trip to the target size runs in pixel space
    after the VAE decode, never on latents; see
    :mod:`fastvideo.mlx_runtime.frame_upsample` for why.
    """
    if scale < 2:
        raise ValueError(f"fast-spatial scale must be at least 2, got {scale}.")
    if upsample_mode not in PIXEL_UPSAMPLE_MODES:
        raise ValueError(f"Unsupported upsample mode: {upsample_mode!r} "
                         f"(expected one of {', '.join(PIXEL_UPSAMPLE_MODES)})")
    if sharpen < 0:
        raise ValueError(f"fast_spatial_sharpen must be non-negative, got {sharpen}.")
    target_canvas_height, target_canvas_width = _model_canvas_size(height, width)
    stage1_height = math.ceil(height / scale)
    stage1_width = math.ceil(width / scale)
    canvas_height, canvas_width = _model_canvas_size(stage1_height, stage1_width)
    if canvas_height * canvas_width >= target_canvas_height * target_canvas_width:
        raise ValueError(
            f"fast-spatial scale {scale} does not reduce the H3 canvas for {height}x{width} "
            f"(stage-1 canvas {canvas_width}x{canvas_height} vs {target_canvas_width}x{target_canvas_height}).")
    return FastSpatialPlan(
        target_height=height,
        target_width=width,
        stage1_height=stage1_height,
        stage1_width=stage1_width,
        canvas_height=canvas_height,
        canvas_width=canvas_width,
        scale=scale,
        upsample_mode=upsample_mode,
        sharpen=sharpen,
    )

fastvideo.mlx_runtime.minimax_h3_pipeline.prompt_cache_path

prompt_cache_path(cache_dir: str | Path, model_root: str | Path, prompt: str) -> Path
Source
def prompt_cache_path(cache_dir: str | Path, model_root: str | Path, prompt: str) -> Path:
    digest = hashlib.sha256(
        f"{MINIMAX_H3_PROMPT_CACHE_VERSION}::{Path(model_root)}::{prompt}".encode()).hexdigest()[:24]
    return Path(cache_dir) / f"prompt_embeds_{digest}.npz"

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline

MiniMaxH3MLXPipeline(*, model_root: str | Path, mlx_dit_checkpoint: str | Path, vae_dtype: str = 'fp32', vae_decoder_path: str | Path | None = None, vae_encoder_path: str | Path | None = None, vae_decoder_quantization: str | None = None, prompt_cache_dir: str | Path | None = None, conditioner_dir: str | Path | None = None, tokenizer_dir: str | Path | None = None, metal_wired_limit_gib: float | None = None, mlx_cache_limit_gib: float | None = None, quant_gemm: str | None = None, video_decode_backend: str = 'h3-vae', taeh3_checkpoint: str | Path | None = None, taeh3_chunk_size: int = 5, conditioner_mode: str = 'auto', resident: bool = False)

Text-to-video-with-audio generation through the native MLX runtime.

generate also takes an ordered references list for the reference-conditioned (Ref2VA / OmniRef) student, which prepends the encoded references to the packed sequence and pins them as conditioning.

Source
class MiniMaxH3MLXPipeline:
    """Text-to-video-with-audio generation through the native MLX runtime.

    ``generate`` also takes an ordered ``references`` list for the
    reference-conditioned (Ref2VA / OmniRef) student, which prepends the
    encoded references to the packed sequence and pins them as conditioning.
    """

    # Defaults for pipelines built without __init__ (unit tests use __new__).
    resident = False
    conditioner_mode = "auto"
    vae_decoder_path: Path | None = None
    vae_encoder_path: Path | None = None
    vae_decoder_quantization: str | None = None

    def __init__(
        self,
        *,
        model_root: str | Path,
        mlx_dit_checkpoint: str | Path,
        vae_dtype: str = "fp32",
        vae_decoder_path: str | Path | None = None,
        vae_encoder_path: str | Path | None = None,
        vae_decoder_quantization: str | None = None,
        prompt_cache_dir: str | Path | None = None,
        conditioner_dir: str | Path | None = None,
        tokenizer_dir: str | Path | None = None,
        metal_wired_limit_gib: float | None = None,
        mlx_cache_limit_gib: float | None = None,
        quant_gemm: str | None = None,
        video_decode_backend: str = "h3-vae",
        taeh3_checkpoint: str | Path | None = None,
        taeh3_chunk_size: int = 5,
        conditioner_mode: str = "auto",
        resident: bool = False,
    ) -> None:
        import mlx.core as mx

        self.model_root = Path(model_root)
        if conditioner_mode not in ("auto", "streamed", "nvfp4"):
            raise ValueError(f"Unknown H3 conditioner mode: {conditioner_mode}")
        if resident and video_decode_backend != "h3-vae":
            raise ValueError("Resident H3 generation requires the H3 video VAE.")
        self.conditioner_mode = conditioner_mode
        self.resident = resident
        self._resident_components: dict[str, Any] = {}
        self.dit_checkpoint = Path(mlx_dit_checkpoint)
        self.vae_dtype = vae_dtype
        self.vae_decoder_path = Path(vae_decoder_path) if vae_decoder_path else None
        self.vae_encoder_path = Path(vae_encoder_path) if vae_encoder_path else None
        self.vae_decoder_quantization = vae_decoder_quantization
        if video_decode_backend not in ("h3-vae", "taeh3"):
            raise ValueError(f"Unknown H3 video decoder: {video_decode_backend}")
        if taeh3_chunk_size < 1:
            raise ValueError("taeh3_chunk_size must be positive.")
        if taeh3_checkpoint is not None and video_decode_backend != "taeh3":
            raise ValueError("taeh3_checkpoint requires video_decode_backend='taeh3'.")
        self.video_decode_backend = video_decode_backend
        self.taeh3_checkpoint = taeh3_checkpoint
        self.taeh3_chunk_size = taeh3_chunk_size
        self.prompt_cache_dir = Path(prompt_cache_dir) if prompt_cache_dir else None
        self.conditioner_dir = Path(conditioner_dir) if conditioner_dir else self.model_root / "text_encoder"
        self.tokenizer_dir = Path(tokenizer_dir) if tokenizer_dir else self.model_root / "tokenizer"
        self._validate_inputs_before_loading()
        manifest = json.loads((self.dit_checkpoint / H3_MANIFEST_FILENAME).read_text())
        dit_config = manifest["config"]
        patch_size = dit_config["patch_size"]
        if len(patch_size) != 3:
            raise ValueError(f"H3 DiT patch_size must have three dimensions, got {patch_size}.")
        self._dit_patch_size = (int(patch_size[0]), int(patch_size[1]), int(patch_size[2]))
        self._dit_in_channels = int(dit_config["in_channels"])
        self.last_dit_forward_s = 0.0
        self.last_vsa_stats: dict[str, Any] | None = None
        self._apply_quant_gemm(quant_gemm)
        self._previous_memory_limit, self._previous_wired_limit = _configure_metal_memory_limits(mx,
                                                                                                 metal_wired_limit_gib,
                                                                                                 resident=resident)
        self._previous_cache_limit = _configure_mlx_cache_limit(mx, mlx_cache_limit_gib)

    def _apply_quant_gemm(self, quant_gemm: str | None) -> None:
        """An explicit mode overrides FASTVIDEO_MLX_QUANT_GEMM for this pipeline; None leaves it in charge."""
        self._quant_gemm_set = quant_gemm is not None
        self._previous_quant_gemm = set_quant_gemm_mode(quant_gemm) if self._quant_gemm_set else None

    def _restore_quant_gemm(self) -> None:
        if getattr(self, "_quant_gemm_set", False):
            set_quant_gemm_mode(self._previous_quant_gemm)
            self._quant_gemm_set = False

    def _load_video_vae(self, *, include_encoder: bool, include_decoder: bool = True):
        """Load the H3 video VAE, swapping in standalone decoder / encoder files when configured."""
        from fastvideo.mlx_runtime import minimax_h3_video_vae as video_vae

        vae_dir = self.model_root / "vae"
        standalone = (self.vae_decoder_path is not None or self.vae_encoder_path is not None
                      or self.vae_decoder_quantization is not None)
        if not standalone:
            return video_vae.mlx_h3_video_vae_from_dir(vae_dir,
                                                       include_encoder=include_encoder,
                                                       include_decoder=include_decoder,
                                                       storage_dtype=self.vae_dtype)
        has_base = vae_dir.is_dir() and (any(vae_dir.glob("*.safetensors")) or (vae_dir / "config.json").exists())
        return video_vae.mlx_h3_video_vae_from_files(
            vae_dir=vae_dir if has_base else None,
            decoder_path=self.vae_decoder_path,
            encoder_path=self.vae_encoder_path,
            include_encoder=include_encoder,
            include_decoder=include_decoder,
            storage_dtype=self.vae_dtype,
            decoder_quantization=self.vae_decoder_quantization,
        )

    # -- input validation (before anything heavy loads) -------------------

    def _validate_inputs_before_loading(self) -> bool:
        missing = []
        if not self.dit_checkpoint.exists():
            missing.append(str(self.dit_checkpoint))
        vae_dir = self.model_root / "vae"
        audio_dir = self.model_root / "audio_vae"
        # Standalone decoder + encoder files (LightVAE, FastH3 decoder) replace the base VAE shards entirely.
        standalone = self.vae_decoder_path is not None and self.vae_encoder_path is not None
        if (self.video_decode_backend == "h3-vae" and not standalone
                and not (vae_dir.exists() and any(vae_dir.glob("*.safetensors")))):
            missing.append(str(vae_dir))
        for path in (self.vae_decoder_path, self.vae_encoder_path):
            if path is not None and not path.is_file():
                missing.append(str(path))
        if not (audio_dir.exists() and any(audio_dir.glob("*.safetensors"))):
            missing.append(str(audio_dir))
        if missing:
            raise FileNotFoundError(f"Missing required H3 components: {missing}")
        return True

    @staticmethod
    def resolve_geometry(
        height: int,
        width: int,
        num_frames: int,
        *,
        enforce_duration: bool = True,
    ) -> dict[str, int]:
        """Explicit canvases pass through (positive multiples of 32); the
        aspect-ratio resolver only applies when dimensions are omitted."""
        if height <= 0 or width <= 0 or height % 32 or width % 32:
            raise ValueError(f"H3 canvas must be positive multiples of 32, got {height}x{width}.")
        aligned_frames = align_num_frames(num_frames)
        latent_frames = video_latent_num_frames(aligned_frames)
        duration = aligned_frames / MINIMAX_H3_FPS
        if enforce_duration and not MINIMAX_H3_MIN_ALIGNED_FRAMES <= aligned_frames <= MINIMAX_H3_MAX_ALIGNED_FRAMES:
            raise ValueError(f"H3 generates {MINIMAX_H3_MIN_DURATION:g}-{MINIMAX_H3_MAX_DURATION:g} s at "
                             f"{MINIMAX_H3_FPS} fps; {aligned_frames} frames is {duration:.2f} s, outside the "
                             f"accepted {MINIMAX_H3_MIN_ALIGNED_FRAMES}-{MINIMAX_H3_MAX_ALIGNED_FRAMES} frame range.")
        return {
            "height": height,
            "width": width,
            "num_frames": aligned_frames,
            "latent_frame_count": latent_frames,
            "latent_height": height // 16,
            "latent_width": width // 16,
        }

    # -- phase 1: conditioning -------------------------------------------

    def prepare_resident(self) -> None:
        """Load the encoder, DiT, and both decoders once, before timed requests."""
        if not self.resident or self._resident_components:
            return
        from fastvideo.mlx_runtime.minimax_h3_conditioner import ResidentNVFP4MiniMaxH3TextConditioner
        from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir
        import mlx.core as mx

        try:
            conditioner = self._load_conditioner()
            if not isinstance(conditioner, ResidentNVFP4MiniMaxH3TextConditioner):
                conditioner.close()
                raise ValueError("All-resident generation requires the packed NVFP4 text encoder.")
            self._resident_components["conditioner"] = conditioner
            self._resident_components["dit"] = _load_resident_dit(self.dit_checkpoint)
            self._resident_components["video_vae"] = self._load_video_vae(include_encoder=False)
            self._resident_components["audio_vae"] = mlx_h3_audio_vae_from_dir(self.model_root / "audio_vae",
                                                                               include_encoder=False)
            mx.eval(list(self._resident_components["audio_vae"].weights.values()))
            logger.info("H3 components resident: %.2f GiB active MLX memory", mx.get_active_memory() / 2**30)
        except Exception:
            self.close()
            raise

    def close(self) -> None:
        self._restore_quant_gemm()
        conditioner = self._resident_components.get("conditioner")
        if conditioner is not None:
            conditioner.close()
        self._resident_components.clear()
        _cleanup_mlx()
        previous = getattr(self, "_previous_wired_limit", None)
        if previous is not None:
            import mlx.core as mx

            _restore_metal_wired_limit(mx, previous)
            self._previous_wired_limit = None
        previous_memory = getattr(self, "_previous_memory_limit", None)
        if previous_memory is not None:
            import mlx.core as mx

            _restore_metal_memory_limit(mx, previous_memory)
            self._previous_memory_limit = None
        previous_cache = getattr(self, "_previous_cache_limit", None)
        if previous_cache is not None:
            import mlx.core as mx

            mx.set_cache_limit(previous_cache)
            self._previous_cache_limit = None

    def encode_prompt(self, prompt: str) -> tuple[np.ndarray, np.ndarray]:
        """Returns (hidden states (S, hidden), token tags). Uses the cache or
        the streamed conditioner."""
        cache_key = self._prompt_cache_key(prompt)
        if cache_key is not None and cache_key.exists():
            data = np.load(cache_key)
            logger.info("Loaded prompt embeddings from cache %s", cache_key)
            return data["hidden_states"], data["token_tags"]

        if self.resident:
            self.prepare_resident()
            conditioner = self._resident_components["conditioner"]
        else:
            conditioner = self._load_conditioner()
        hidden, tags = conditioner.encode_prompt(prompt)
        if not self.resident:
            conditioner.close()
        _cleanup_mlx()
        if cache_key is not None:
            cache_key.parent.mkdir(parents=True, exist_ok=True)
            tmp_cache = cache_key.with_name(f".{cache_key.name}.tmp")
            try:
                with tmp_cache.open("wb") as handle:
                    np.savez(handle, hidden_states=hidden, token_tags=tags)
                tmp_cache.replace(cache_key)
            finally:
                tmp_cache.unlink(missing_ok=True)
        return hidden, tags

    def _prompt_cache_key(self, prompt: str) -> Path | None:
        """Cache path bound to the effective encoder and its files; None skips the cache."""
        if self.prompt_cache_dir is None:
            return None
        fingerprint = {
            "conditioner": self._conditioner_kind(),
            "text_encoder": text_encoder_fingerprint(self.conditioner_dir),
            "tokenizer": text_encoder_fingerprint(self.tokenizer_dir),
        }
        if not (fingerprint["text_encoder"]["complete"] and fingerprint["tokenizer"]["complete"]):
            return None
        return prompt_cache_path(self.prompt_cache_dir, fingerprint_digest(fingerprint), prompt)

    def load_prompt_cache(self, path: str | Path) -> tuple[np.ndarray, np.ndarray]:
        data = np.load(path)
        return data["hidden_states"], data["token_tags"]

    def prepare_references(
        self,
        references: Sequence[MiniMaxH3Reference],
        num_frames: int,
        *,
        max_pixels: int | None = None,
    ) -> list[MiniMaxH3PreparedReference]:
        """Validate and decode the ordered reference list.

        Decoding happens once, before anything is loaded, so a bad reference
        fails before the DiT and the encoders are resident.
        """
        validated = validate_references(list(references))
        target_sample_rate = self.audio_sample_rate()
        prepared = [
            prepare_reference(reference, num_frames, target_sample_rate, max_pixels=max_pixels)
            for reference in validated
        ]
        logger.info("Prepared %d reference(s): %s", len(prepared),
                    ", ".join(reference.media_type for reference in prepared))
        return prepared

    def audio_sample_rate(self) -> int:
        """The audio VAE's sample rate, which reference soundtracks are resampled to."""
        config_path = self.model_root / "audio_vae" / "config.json"
        if not config_path.is_file():
            return 32000
        return int(json.loads(config_path.read_text(encoding="utf-8")).get("sampling_rate", 32000))

    def encode_reference_prompt(
        self,
        prompt: str,
        references: list[MiniMaxH3PreparedReference],
    ) -> tuple[np.ndarray, np.ndarray]:
        """Encode the prompt with its ordered references as Qwen3-VL condition rows."""
        from fastvideo.mlx_runtime.minimax_h3_ref_conditioning import encode_reference_prompt, load_vision_limits
        from fastvideo.mlx_runtime.minimax_h3_vision import Qwen3VLVisionConfig

        if self.resident:
            self.prepare_resident()
            conditioner = self._resident_components["conditioner"]
        else:
            conditioner = self._load_conditioner()
        try:
            config = Qwen3VLVisionConfig.from_config_json(self.conditioner_dir / "config.json")
            min_pixels, max_pixels = load_vision_limits(self.processor_dir())
            video_min_pixels, video_max_pixels = load_vision_limits(self.processor_dir(), video=True)
            hidden, tags = encode_reference_prompt(
                conditioner,
                prompt,
                references,
                vision_config=config,
                min_pixels=min_pixels,
                max_pixels=max_pixels,
                video_min_pixels=video_min_pixels,
                video_max_pixels=video_max_pixels,
            )
        finally:
            if not self.resident:
                conditioner.close()
        _cleanup_mlx()
        return hidden, tags

    def processor_dir(self) -> Path:
        """Where the Qwen3-VL processor config lives, defaulting to the tokenizer dir."""
        for candidate in (self.model_root / "processor", self.tokenizer_dir):
            if (candidate / "preprocessor_config.json").is_file():
                return candidate
        return self.tokenizer_dir

    def has_conditioner_weights(self) -> bool:
        marker = self.conditioner_dir / "model.safetensors.index.json"
        single = self.conditioner_dir / "model.safetensors"
        return marker.exists() or single.exists()

    def _load_conditioner(self):
        from fastvideo.mlx_runtime.minimax_h3_conditioner import (
            ResidentNVFP4MiniMaxH3TextConditioner,
            StreamedMiniMaxH3TextConditioner,
        )

        if self._conditioner_kind() == "nvfp4":
            return ResidentNVFP4MiniMaxH3TextConditioner(self.conditioner_dir, self.tokenizer_dir)
        return StreamedMiniMaxH3TextConditioner(self.conditioner_dir, self.tokenizer_dir)

    def _conditioner_kind(self) -> str:
        """The encoder that conditioner_mode selects for these weights: 'nvfp4' or 'streamed'."""
        config = json.loads((self.conditioner_dir / "config.json").read_text())
        packed = str(config.get("quantization_config", {}).get("quant_method", "")).lower() == "nvfp4"
        if self.conditioner_mode == "nvfp4" or (self.conditioner_mode == "auto" and packed):
            return "nvfp4"
        if packed:
            raise ValueError("The streamed conditioner requires BF16 weights; use conditioner_mode='nvfp4'.")
        return "streamed"

    # -- phase 2: denoise --------------------------------------------------

    def denoise(
        self,
        text_rows: np.ndarray,
        token_tags: np.ndarray,
        *,
        height: int,
        width: int,
        num_frames: int,
        audio_num_frames: int | None = None,
        video_temporal_scale: float = 1.0,
        seed: int,
        num_steps: int = 4,
        dit: Any | None = None,
        vsa_config: MiniMaxH3VSAConfig | None = None,
        inter_step_cooldown_s: float = 0.0,
        references: list[MiniMaxH3PreparedReference] | None = None,
        condition_video_rows: np.ndarray | None = None,
        condition_audio_rows: np.ndarray | None = None,
        reuse_prefix_kv: bool = False,
        prefix_kv_refresh_every: int = 0,
        prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
        prefix_kv_storage: str = "bf16",
    ) -> tuple[np.ndarray, np.ndarray]:
        """Denoise joint latents; returns (normalized target video rows, target audio rows).

        With ``references`` the packed sequence is reference-prefixed and the
        encoder rows ride along in every forward, but only the target rows are
        stepped and returned: the references are conditioning, not output.

        ``reuse_prefix_kv`` (Ref2VA only, lossy) records the text/reference
        rows' per-block K/V on the first step and runs later steps on the
        target rows alone; ``prefix_kv_refresh_every=R`` re-records every R
        steps (0 never). The cache is refused above ``prefix_kv_cache_max_gib``.
        """
        import mlx.core as mx

        self.last_prefix_kv_stats: dict[str, object] = {"enabled": False}
        self.last_denoise_step_times = []
        if prefix_kv_refresh_every < 0:
            raise ValueError(f"prefix_kv_refresh_every must be >= 0, got {prefix_kv_refresh_every}.")
        if reuse_prefix_kv and references is None:
            raise ValueError("Prefix K/V reuse needs Ref2VA references (a reference-prefixed packed layout).")
        validate_prefix_kv_storage(prefix_kv_storage)

        geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=audio_num_frames is None)
        audio_frames = geometry["num_frames"] if audio_num_frames is None else align_num_frames(audio_num_frames)

        resident_dit = dit is None and self.resident
        if resident_dit:
            _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
            self.prepare_resident()
            dit = self._resident_components["dit"]
        owned_dit = dit is None
        if owned_dit:
            _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
            t0 = time.perf_counter()
            dit = load_mlx_h3_checkpoint(self.dit_checkpoint)
            logger.info("Loaded MLX H3 DiT from %s in %.1fs", self.dit_checkpoint, time.perf_counter() - t0)

        schedule = resolve_h3_denoise_schedule(self.model_root, num_steps)
        video_scheduler = schedule.video
        audio_scheduler = schedule.audio
        union = schedule.adaln_timesteps
        num_steps = schedule.num_steps
        # A FastH3 PDD export runs one fused head block per trained block; its
        # plan supplies the block partition, the per-modality node sigmas
        # (already on the schedulers) and the integration weights.
        pdd_plan = schedule.pdd_plan
        # The keyframe-noise timestep (0.999) is only exercised by FL2VA/Ref2VA
        # conditioning rows. A Ref2VA PDD ladder already carries it, so a
        # converted DiT's baked cache matches; this append is then a no-op.
        if references is not None:
            if video_temporal_scale != 1.0:
                # The reference-prefixed builder stretches only the target video
                # rows, which the CUDA packing never does either; refusing beats
                # silently ignoring the request.
                raise ValueError("Reference-conditioned packing has no temporal-scale stretch; "
                                 "run Ref2VA without fast mode.")
            union = np.unique(np.concatenate([union, np.array([MINIMAX_H3_KEYFRAME_NOISE_AUG],
                                                              dtype=np.float32)])).astype(np.float32)
            if condition_video_rows is None or condition_audio_rows is None:
                condition_video_rows, condition_audio_rows = self._encode_reference_conditions(
                    references, dit.patch_size, video_scheduler, seed)

        cache = getattr(dit, "_adaln_cache", None)
        if cache is None or not np.array_equal(cache.timesteps.astype(np.float32), union):
            if cache is not None:
                extra = np.setdiff1d(union, cache.timesteps)
                logger.info("Recomputing AdaLN cache for %d-step ladder (extra timesteps %s).", num_steps, extra)
            if _adaln_weights_dropped(dit):
                # An earlier request on this DiT already released the AdaLN projections.
                if not resident_dit:
                    raise ValueError("This H3 DiT dropped its AdaLN weights for another step ladder; "
                                     "pass a freshly loaded DiT for a different num_steps.")
                logger.info("Reloading the resident H3 DiT for the %d-step ladder.", num_steps)
                cache = dit = None
                self._resident_components.pop("dit", None)
                _cleanup_mlx()
                dit = self._resident_components["dit"] = _load_resident_dit(self.dit_checkpoint)
            dit.precompute_adaln(union, drop_weights=True)

        if resident_dit:
            # The resident DiT keeps the previous request's VSA mode unless reset to the default.
            vsa_config = vsa_config or MiniMaxH3VSAConfig()
        if vsa_config is not None:
            dit.configure_vsa(vsa_config)
        if hasattr(dit, "reset_vsa_stats"):
            dit.reset_vsa_stats()

        if references is not None:
            layout = build_ref2va_packed_layout(
                np.asarray(token_tags, dtype=np.int64),
                references,
                geometry["latent_frame_count"],
                geometry["latent_height"],
                geometry["latent_width"],
                audio_latent_num_frames(audio_frames),
                patch_size=dit.patch_size,
            )
        else:
            layout = build_packed_layout(
                len(token_tags),
                geometry["latent_frame_count"],
                geometry["latent_height"],
                geometry["latent_width"],
                audio_latent_num_frames(audio_frames),
                patch_size=dit.patch_size,
                text_token_tags=np.asarray(token_tags, dtype=np.int64),
                video_temporal_scale=video_temporal_scale,
            )
        if getattr(dit, "vsa_config", None) is not None and dit.vsa_config.enabled:
            dit.prepare_vsa_geometry(layout)

        video_key, audio_key = mx.random.split(mx.random.key(seed))
        target_video_rows = int(layout.video_indices.shape[0] - layout.num_condition_video_rows)
        target_audio_rows = int(layout.audio_indices.shape[0] - layout.num_condition_audio_rows)
        x_v = mx.random.normal((target_video_rows, dit.patch_dim), key=video_key)
        x_a = mx.random.normal((target_audio_rows, dit.audio_in_channels), key=audio_key)
        if condition_video_rows is not None:
            if condition_video_rows.shape != (layout.num_condition_video_rows, dit.patch_dim):
                raise ValueError(f"Video condition rows {condition_video_rows.shape} do not match the layout's "
                                 f"{layout.num_condition_video_rows} condition rows of width {dit.patch_dim}.")
            x_v = mx.concatenate([mx.array(condition_video_rows), x_v], axis=0)
        elif layout.num_condition_video_rows:
            raise ValueError("A reference-prefixed layout needs its encoded video condition rows.")
        if condition_audio_rows is not None:
            if condition_audio_rows.shape != (layout.num_condition_audio_rows, dit.audio_in_channels):
                raise ValueError(f"Audio condition rows {condition_audio_rows.shape} do not match the layout's "
                                 f"{layout.num_condition_audio_rows} condition rows.")
            x_a = mx.concatenate([mx.array(condition_audio_rows), x_a], axis=0)
        elif layout.num_condition_audio_rows:
            raise ValueError("A reference-prefixed layout needs its encoded audio condition rows.")
        text = mx.array(text_rows.astype(np.float32))
        dit_forward_s = 0.0
        prefix_kv = None
        if reuse_prefix_kv:
            frozen_rows = frozen_prefix_rows(layout)
            estimated = estimate_prefix_kv_bytes(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                                 prefix_kv_storage)
            check_prefix_kv_budget(estimated, prefix_kv_cache_max_gib, rows=frozen_rows)
            prefix_kv = MiniMaxH3PrefixKVCache(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                               prefix_kv_storage)
            logger.info("Prefix K/V reuse: %d frozen rows, ~%.2f GiB cache (%s), refresh_every=%d", frozen_rows,
                        estimated / 2**30, prefix_kv_storage, prefix_kv_refresh_every)
        recorded_steps: list[int] = []
        reused_steps: list[int] = []

        if pdd_plan is not None:
            # PDD targets start at the first node's noise level (0.999 * eps),
            # as the students were trained; conditioning rows stay untouched.
            x_v = mx.concatenate([
                x_v[:layout.num_condition_video_rows],
                x_v[layout.num_condition_video_rows:] * float(pdd_plan.node_sigmas["video"][0]),
            ],
                                 axis=0)
            x_a = mx.concatenate([
                x_a[:layout.num_condition_audio_rows],
                x_a[layout.num_condition_audio_rows:] * float(pdd_plan.node_sigmas["audio"][0]),
            ],
                                 axis=0)

        for step_index in range(num_steps):
            video_t = float(video_scheduler.timesteps[step_index])
            audio_t = float(audio_scheduler.timesteps[step_index])
            unique, inverse = build_row_timesteps(
                layout,
                video_timestep=video_t,
                audio_timestep=audio_t,
                condition_video_timestep=max(video_t, MINIMAX_H3_KEYFRAME_NOISE_AUG),
                condition_audio_timestep=1.0,
            )
            step_started = time.perf_counter()
            pdd_forward_kwargs = ({} if pdd_plan is None else {
                "pdd_block": pdd_plan.block(step_index),
                "pdd_integration_weights": pdd_plan.integration_weights,
            })
            record_step = prefix_kv is not None and (step_index == 0 or (prefix_kv_refresh_every > 0
                                                                         and step_index % prefix_kv_refresh_every == 0))
            if prefix_kv is not None and not record_step:
                video_velocity, audio_velocity = dit.forward_target_rows(
                    x_v[layout.num_condition_video_rows:],
                    x_a[layout.num_condition_audio_rows:],
                    layout=layout,
                    step_timesteps=unique,
                    row_timestep_inverse=inverse,
                    prefix_kv=prefix_kv,
                    step_index=step_index,
                    **pdd_forward_kwargs,
                )
                reused_steps.append(step_index)
            else:
                if record_step:
                    pdd_forward_kwargs["record_prefix_kv"] = prefix_kv
                    recorded_steps.append(step_index)
                video_velocity, audio_velocity = dit.forward_with_cache(
                    x_v,
                    x_a,
                    text,
                    layout=layout,
                    step_timesteps=unique,
                    row_timestep_inverse=inverse,
                    step_index=step_index,
                    **pdd_forward_kwargs,
                )
                video_velocity = video_velocity[layout.num_condition_video_rows:]
                audio_velocity = audio_velocity[layout.num_condition_audio_rows:]
            # Only the target rows are denoised: condition rows stay pinned at
            # the timestep the contract trained them at.
            if layout.num_condition_video_rows:
                x_v = mx.concatenate([
                    x_v[:layout.num_condition_video_rows],
                    video_scheduler.step(video_velocity, step_index, x_v[layout.num_condition_video_rows:])
                ],
                                     axis=0)
            else:
                x_v = video_scheduler.step(video_velocity, step_index, x_v)
            if layout.num_condition_audio_rows:
                x_a = mx.concatenate([
                    x_a[:layout.num_condition_audio_rows],
                    audio_scheduler.step(audio_velocity, step_index, x_a[layout.num_condition_audio_rows:])
                ],
                                     axis=0)
            else:
                x_a = audio_scheduler.step(audio_velocity, step_index, x_a)
            mx.eval(x_v, x_a)
            step_s = time.perf_counter() - step_started
            dit_forward_s += step_s
            self.last_denoise_step_times.append(step_s)
            logger.info("H3 denoise step %d/%d (sigma_video=%.4f, sigma_audio=%.4f) in %.2fs", step_index + 1,
                        num_steps, 1.0 - video_t, 1.0 - audio_t, step_s)
            del video_velocity, audio_velocity
            _cleanup_mlx()
            if inter_step_cooldown_s > 0 and step_index < num_steps - 1:
                logger.info("Thermal cooldown %.1fs before step %d...", inter_step_cooldown_s, step_index + 2)
                time.sleep(inter_step_cooldown_s)

        self.last_dit_forward_s = dit_forward_s
        self.last_prefix_kv_stats = {
            "enabled": prefix_kv is not None,
            "refresh_every": prefix_kv_refresh_every,
            "storage": prefix_kv_storage,
            "frozen_rows": 0 if prefix_kv is None else prefix_kv.rows,
            "cache_gib": 0.0 if prefix_kv is None else prefix_kv.nbytes / 2**30,
            "max_gib": prefix_kv_cache_max_gib,
            "recorded_steps": recorded_steps,
            "reused_steps": reused_steps,
        }
        prefix_kv = None
        stats = getattr(dit, "last_vsa_stats", None)
        self.last_vsa_stats = None if stats is None else {
            "enabled": bool(getattr(dit.vsa_config, "enabled", False)),
            "configured_sparsity": stats.configured_sparsity,
            "layer_sparsity": stats.layer_sparsity,
            "achieved_sparsity": stats.achieved_sparsity,
            "tile_size": stats.tile_size,
            "prefix_mode": stats.prefix_mode,
            "impl": stats.impl,
            "num_prefix_tiles": stats.num_prefix_tiles,
            "num_video_tiles": stats.num_video_tiles,
            "video_keep": stats.video_keep,
            "dense_fallback_reason": stats.dense_fallback_reason,
            "attention_calls": stats.attention_calls,
            "sparse_calls": stats.sparse_calls,
            "impl_counts": stats.impl_counts,
            "fallback_reasons": stats.fallback_reasons,
            "checkpoint_vsa_capable": bool(getattr(dit, "vsa_capable", False)),
        }

        video_rows = np.asarray(x_v, dtype=np.float32)[layout.num_condition_video_rows:]
        audio_rows = np.asarray(x_a, dtype=np.float32)[layout.num_condition_audio_rows:]
        if owned_dit:
            del dit
            _cleanup_mlx()
        return video_rows, audio_rows

    # -- phase 3a: video decode -------------------------------------------

    def decode_video(self,
                     video_rows: np.ndarray,
                     *,
                     height: int,
                     width: int,
                     num_frames: int,
                     tiled: bool = True,
                     vae_tile_height: int = 256,
                     vae_tile_width: int = 256) -> np.ndarray:
        """Normalized packed rows -> (T, H, W, 3) uint8 frames."""
        _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
        import mlx.core as mx

        geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=False)
        if self.video_decode_backend == "taeh3":
            from fastvideo.mlx_runtime.minimax_h3_taeh3 import decode_latents_taeh3_mlx

            if self._dit_in_channels != 24:
                raise ValueError("TAEH3 requires a 24-channel H3 checkpoint.")
            latents = unpatchify_video_tokens(video_rows, geometry["latent_frame_count"], geometry["latent_height"],
                                              geometry["latent_width"], self._dit_in_channels, self._dit_patch_size)
            pixels = decode_latents_taeh3_mlx(latents,
                                              checkpoint_path=self.taeh3_checkpoint,
                                              dtype=self.vae_dtype,
                                              chunk_size=self.taeh3_chunk_size)
            frames = (pixels[0] * 255.0).astype(np.uint8)
            if frames.shape != (geometry["num_frames"], height, width, 3):
                raise RuntimeError(f"TAEH3 produced unexpected frame shape: {frames.shape}")
            _cleanup_mlx()
            return frames
        if self.resident:
            self.prepare_resident()
            vae = self._resident_components["video_vae"]
        else:
            vae = self._load_video_vae(include_encoder=False)
        expected_height = height // vae.spatial_compression_ratio
        expected_width = width // vae.spatial_compression_ratio
        if (geometry["latent_height"], geometry["latent_width"]) != (expected_height, expected_width):
            raise RuntimeError("H3 pipeline/VAE spatial compression mismatch: "
                               f"pipeline={(geometry['latent_height'], geometry['latent_width'])}, "
                               f"VAE={(expected_height, expected_width)}.")
        if vae.latent_channels != self._dit_in_channels:
            raise RuntimeError(
                f"H3 DiT/VAE latent-channel mismatch: DiT={self._dit_in_channels}, VAE={vae.latent_channels}.")
        latents = unpatchify_video_tokens(
            video_rows,
            geometry["latent_frame_count"],
            geometry["latent_height"],
            geometry["latent_width"],
            vae.latent_channels,
            self._dit_patch_size,
        )
        z = mx.array(latents)
        z = vae.denormalize_latents(z)
        decoded = vae.decode(z,
                             tiled=tiled,
                             tile_sample_min_height=min(geometry["height"], vae_tile_height),
                             tile_sample_min_width=min(geometry["width"], vae_tile_width))
        pixels = np.clip(np.asarray(vae.denormalize_pixels(decoded)), 0.0, 1.0)
        del vae, decoded, z
        _cleanup_mlx()
        frames = (pixels[0].transpose(1, 2, 3, 0) * 255.0).astype(np.uint8)  # (T, H, W, C)
        if frames.shape[0] != geometry["num_frames"]:
            raise RuntimeError(f"decoded {frames.shape[0]} frames, expected {geometry['num_frames']}")
        return frames

    # -- phase 3b: audio decode --------------------------------------------

    def decode_audio(self, audio_rows: np.ndarray, *, num_frames: int) -> np.ndarray:
        """Normalized packed audio rows -> stereo waveform (2, S) fp32 in [-1, 1]."""
        import mlx.core as mx

        from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir

        num_audio_latents = audio_latent_num_frames(align_num_frames(num_frames))
        latents = unpack_audio_tokens(audio_rows, num_audio_latents)
        if self.resident:
            self.prepare_resident()
            vae = self._resident_components["audio_vae"]
        else:
            vae = mlx_h3_audio_vae_from_dir(self.model_root / "audio_vae", include_encoder=False)
        z = vae.denormalize_latents(mx.array(latents))
        waveform = np.asarray(vae.decode(z))[:, 0, :]  # (B, 1, S) -> (B, S)
        del vae, z
        _cleanup_mlx()
        # Keep audio at least as long as the final video packet. Rounding down
        # by a fractional sample makes ffmpeg's ``-shortest`` drop frame 124.
        expected_samples = _audio_sample_count(align_num_frames(num_frames))
        if waveform.shape[-1] < expected_samples:
            waveform = np.pad(waveform, ((0, 0), (0, expected_samples - waveform.shape[-1])))
        return np.clip(waveform[:, :expected_samples], -1.0, 1.0)

    # -- phase 4: mux --------------------------------------------------------

    def mux(self,
            frames: np.ndarray,
            waveform: np.ndarray,
            output_path: str | Path,
            fps: int = MINIMAX_H3_FPS,
            sample_rate: int = 32000) -> Path:
        """H.264 video + AAC stereo audio, A/V durations within one frame."""
        output_path = Path(output_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        tmp_video = output_path.with_suffix(".tmp.mp4")
        tmp_audio = output_path.with_suffix(".tmp.wav")

        pcm = (np.clip(waveform.T, -1.0, 1.0) * 32767.0).astype("<i2")  # (S, 2)
        import wave

        try:
            with wave.open(str(tmp_audio), "wb") as handle:
                handle.setnchannels(2)
                handle.setsampwidth(2)
                handle.setframerate(sample_rate)
                handle.writeframes(pcm.tobytes())

            height, width = frames.shape[1:3]
            ffmpeg = shutil.which("ffmpeg")
            if ffmpeg is None:
                raise RuntimeError("ffmpeg is required for MP4 muxing.")
            subprocess.run(
                [
                    ffmpeg,
                    "-y",
                    "-loglevel",
                    "error",
                    "-f",
                    "rawvideo",
                    "-pix_fmt",
                    "rgb24",
                    "-s",
                    f"{width}x{height}",
                    "-r",
                    str(fps),
                    "-i",
                    "-",
                    "-i",
                    str(tmp_audio),
                    "-c:v",
                    "libx264",
                    "-preset",
                    "medium",
                    "-crf",
                    "18",
                    "-pix_fmt",
                    "yuv420p",
                    "-c:a",
                    "aac",
                    "-b:a",
                    "192k",
                    "-shortest",
                    "-movflags",
                    "+faststart",
                    str(tmp_video),
                ],
                input=frames.tobytes(),
                check=True,
            )
            tmp_video.replace(output_path)
        finally:
            tmp_audio.unlink(missing_ok=True)
            tmp_video.unlink(missing_ok=True)
        return output_path

    def _encode_reference_conditions(
        self,
        references: list[MiniMaxH3PreparedReference],
        patch_size: tuple[int, int, int],
        scheduler: MiniMaxH3SchedulerState,
        seed: int,
    ) -> tuple[np.ndarray, np.ndarray | None]:
        """VAE-encode the prepared references into their condition rows.

        The encoders are loaded for this call alone and released afterwards
        rather than kept on the pipeline: only a reference-conditioned request
        needs them, and they double the VAE's resident bytes.
        """
        from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir
        from fastvideo.mlx_runtime.minimax_h3_ref_conditioning import encode_reference_conditions

        if self.video_decode_backend == "taeh3":
            raise ValueError("A reference-conditioned request needs the video VAE encoder; "
                             "use --video-decode-backend h3-vae.")

        needs_audio = any(reference.has_audio for reference in references)
        video_vae = self._load_video_vae(include_encoder=True, include_decoder=False)
        audio_vae = (mlx_h3_audio_vae_from_dir(self.model_root /
                                               "audio_vae", include_encoder=True) if needs_audio else None)
        try:
            return encode_reference_conditions(
                video_vae=video_vae,
                audio_vae=audio_vae,
                references=references,
                patch_size=patch_size,
                scheduler=scheduler,
                seed=seed,
            )
        finally:
            del video_vae, audio_vae
            _cleanup_mlx()

    # -- end-to-end ----------------------------------------------------------

    def generate(
        self,
        prompt: str,
        *,
        output_path: str | Path,
        references: Sequence[MiniMaxH3Reference] | None = None,
        height: int = 480,
        width: int = 832,
        num_frames: int = 124,
        seed: int = 0,
        num_steps: int = 4,
        save_frames: bool = False,
        tiled_video_decode: bool = True,
        vae_tile_height: int = 256,
        vae_tile_width: int = 256,
        fast: bool = False,
        fast_factor: int = 2,
        fast_sharpen: float = 0.6,
        rife_weights_dir: str | Path | None = None,
        fast_spatial: bool = False,
        fast_spatial_scale: int = 2,
        fast_spatial_upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE,
        fast_spatial_sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN,
        vsa: bool | None = None,
        vsa_sparsity: float | None = None,
        vsa_tile_size: int | None = None,
        vsa_ref_policy: str | None = None,
        vsa_ref_keep_rate: float | None = None,
        vsa_prefix_mode: str = "exempt",
        vsa_dense_first_n_steps: int = 0,
        vsa_dense_layers: tuple[int, ...] = (),
        vsa_impl: str = "auto",
        inter_step_cooldown_s: float = 0.0,
        reference_max_pixels: int | None = None,
        reuse_prefix_kv: bool = False,
        prefix_kv_refresh_every: int = 0,
        prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
        prefix_kv_storage: str = "bf16",
    ) -> GenerationResult:
        timings: dict[str, float] = {}
        peaks: dict[str, float] = {}

        if fast_sharpen < 0:
            raise ValueError(f"fast_sharpen must be non-negative, got {fast_sharpen}.")
        if inter_step_cooldown_s < 0:
            raise ValueError(f"inter_step_cooldown_s must be non-negative, got {inter_step_cooldown_s}.")
        _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
        _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
        # Unset VSA knobs come from the snapshot's FastH3 contract; ``vsa=None``
        # turns VSA on for a VSA contract and ``vsa=False`` opts out.
        vsa_settings = resolve_vsa_settings(
            load_fasth3_inference_contract(self.model_root),
            vsa=vsa,
            sparsity=vsa_sparsity,
            tile_size=vsa_tile_size,
            ref_policy=vsa_ref_policy,
            ref_keep_rate=vsa_ref_keep_rate,
            checkpoint_capable=mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
        )
        vsa = vsa_settings.enabled
        vsa_config = MiniMaxH3VSAConfig(
            enabled=vsa,
            sparsity=vsa_settings.sparsity,
            tile_size=vsa_settings.tile_size,
            ref_policy=vsa_settings.ref_policy,
            ref_keep_rate=vsa_settings.ref_keep_rate,
            prefix_mode=vsa_prefix_mode,  # type: ignore[arg-type]
            dense_first_n_steps=vsa_dense_first_n_steps,
            dense_layers=vsa_dense_layers,
            impl=vsa_impl,  # type: ignore[arg-type]
        ) if vsa else MiniMaxH3VSAConfig()
        if vsa_config.enabled and not mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint):
            raise dense_only_vsa_error(self.dit_checkpoint)
        spatial_plan = plan_fast_spatial(
            height,
            width,
            scale=fast_spatial_scale,
            upsample_mode=fast_spatial_upsample_mode,
            sharpen=fast_spatial_sharpen,
        ) if fast_spatial else None
        _preflight_media_dependencies(
            fast=fast,
            fast_sharpen=fast_sharpen,
            rife_weights_dir=rife_weights_dir,
            fast_spatial=fast_spatial,
        )
        if spatial_plan is not None:
            canvas_height, canvas_width = spatial_plan.canvas_height, spatial_plan.canvas_width
        else:
            canvas_height, canvas_width = _model_canvas_size(height, width)
        target_geometry = self.resolve_geometry(canvas_height, canvas_width, num_frames)
        fast_plan = plan_fast_temporal(target_geometry["num_frames"], fast_factor) if fast else None
        video_num_frames = fast_plan.source_frames if fast_plan is not None else target_geometry["num_frames"]
        video_temporal_scale = fast_plan.video_temporal_scale if fast_plan is not None else 1.0
        video_geometry = self.resolve_geometry(
            canvas_height,
            canvas_width,
            video_num_frames,
            enforce_duration=fast_plan is None,
        )
        logger.info(
            "Geometry: output=%dx%dx%d model=%dx%dx%d audio_frames=%d fast=%s fast_spatial=%s",
            width,
            height,
            target_geometry["num_frames"],
            canvas_width,
            canvas_height,
            video_geometry["num_frames"],
            target_geometry["num_frames"],
            fast_plan,
            spatial_plan,
        )

        if self.video_decode_backend == "taeh3":
            from fastvideo.mlx_runtime.minimax_h3_taeh3 import ensure_taeh3_checkpoint

            started = time.perf_counter()
            ensure_taeh3_checkpoint(self.taeh3_checkpoint)
            timings["decoder_prepare_s"] = time.perf_counter() - started
            logger.warning("TAEH3 is an approximate preview decoder; reconstruction differs from the full H3 VAE.")

        prepared_references: list[MiniMaxH3PreparedReference] | None = None
        if references is not None:
            prepared_references = self.prepare_references(references,
                                                          video_geometry["num_frames"],
                                                          max_pixels=reference_max_pixels)

        _reset_peak_memory()
        started = time.perf_counter()
        if prepared_references is None:
            text_rows, token_tags = self.encode_prompt(prompt)
        else:
            text_rows, token_tags = self.encode_reference_prompt(prompt, prepared_references)
        timings["condition_s"] = time.perf_counter() - started
        peaks["condition_gib"] = _peak_memory_gib()
        _cleanup_mlx()

        _reset_peak_memory()
        started = time.perf_counter()
        video_rows, audio_rows = self.denoise(
            text_rows,
            token_tags,
            height=video_geometry["height"],
            width=video_geometry["width"],
            num_frames=video_geometry["num_frames"],
            audio_num_frames=target_geometry["num_frames"] if fast_plan is not None else None,
            video_temporal_scale=video_temporal_scale,
            seed=seed,
            num_steps=num_steps,
            vsa_config=vsa_config,
            inter_step_cooldown_s=inter_step_cooldown_s,
            references=prepared_references,
            reuse_prefix_kv=reuse_prefix_kv,
            prefix_kv_refresh_every=prefix_kv_refresh_every,
            prefix_kv_cache_max_gib=prefix_kv_cache_max_gib,
            prefix_kv_storage=prefix_kv_storage,
        )
        timings["denoise_s"] = time.perf_counter() - started
        timings["dit_forward_s"] = float(getattr(self, "last_dit_forward_s", 0.0))
        peaks["denoise_gib"] = _peak_memory_gib()
        del text_rows
        _cleanup_mlx()

        _reset_peak_memory()
        started = time.perf_counter()
        frames = self.decode_video(
            video_rows,
            height=video_geometry["height"],
            width=video_geometry["width"],
            num_frames=video_geometry["num_frames"],
            tiled=tiled_video_decode,
            vae_tile_height=vae_tile_height,
            vae_tile_width=vae_tile_width,
        )
        if spatial_plan is not None:
            frames = _center_crop_frames(frames, spatial_plan.stage1_height, spatial_plan.stage1_width)
        else:
            frames = _center_crop_frames(frames, height, width)
        timings["video_decode_s"] = time.perf_counter() - started
        peaks["video_decode_gib"] = _peak_memory_gib()
        _cleanup_mlx()

        if fast_plan is not None:
            from fastvideo.mlx_runtime.rife_interp import interpolate_to_frame_count, load_model

            _reset_peak_memory()
            started = time.perf_counter()
            model = load_model(weights_dir=str(rife_weights_dir) if rife_weights_dir is not None else None)
            try:
                interpolated = interpolate_to_frame_count(
                    frames,
                    target_geometry["num_frames"],
                    model=model,
                )
                if spatial_plan is None:
                    interpolated = _sharpen_frames(interpolated, fast_sharpen)
                frames = np.stack(interpolated)
                if frames.shape[0] != target_geometry["num_frames"]:
                    raise RuntimeError(
                        f"RIFE produced {frames.shape[0]} frames, expected {target_geometry['num_frames']}.")
                del interpolated
                timings["rife_s"] = time.perf_counter() - started
                peaks["rife_gib"] = _peak_memory_gib()
            finally:
                load_model.cache_clear()
                del model
                _cleanup_mlx()

        if spatial_plan is not None:
            started = time.perf_counter()
            # One sharpen pass, at full resolution: RIFE and the resample soften
            # for the same reason, so the stronger requested amount is applied
            # once instead of stacking two unsharp masks.
            sharpen = spatial_plan.sharpen if fast_plan is None else max(spatial_plan.sharpen, fast_sharpen)
            frames = np.stack(
                upsample_frames(
                    frames,
                    width=spatial_plan.target_width,
                    height=spatial_plan.target_height,
                    mode=spatial_plan.upsample_mode,
                    sharpen=sharpen,
                ))
            timings["spatial_upsample_s"] = time.perf_counter() - started

        _reset_peak_memory()
        started = time.perf_counter()
        waveform = self.decode_audio(audio_rows, num_frames=target_geometry["num_frames"])
        timings["audio_decode_s"] = time.perf_counter() - started
        peaks["audio_decode_gib"] = _peak_memory_gib()
        _cleanup_mlx()

        started = time.perf_counter()
        video_path = self.mux(frames, waveform, output_path)
        timings["mux_s"] = time.perf_counter() - started
        timings["generate_s"] = sum(
            timings.get(key, 0.0) for key in ("decoder_prepare_s", "condition_s", "denoise_s", "video_decode_s",
                                              "rife_s", "spatial_upsample_s", "audio_decode_s", "mux_s"))

        result = GenerationResult(
            video_path=str(video_path),
            frames=frames if save_frames else None,
            waveform=waveform,
            sample_rate=32000,
            timings=timings,
            peak_memory_gib=peaks,
            video_decode_backend=self.video_decode_backend,
            prefix_kv={
                **getattr(self, "last_prefix_kv_stats", {"enabled": False}),
                "step_times_s":
                list(getattr(self, "last_denoise_step_times", [])),
            },
            vsa=getattr(self, "last_vsa_stats", None) or {
                "enabled": vsa_config.enabled,
                "checkpoint_vsa_capable": mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
            },
        )
        logger.info("Generation complete: %s | timings=%s peaks=%s", video_path, timings, peaks)
        return result

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.resolve_geometry

resolve_geometry(height: int, width: int, num_frames: int, *, enforce_duration: bool = True) -> dict[str, int]

Explicit canvases pass through (positive multiples of 32); the aspect-ratio resolver only applies when dimensions are omitted.

Source
@staticmethod
def resolve_geometry(
    height: int,
    width: int,
    num_frames: int,
    *,
    enforce_duration: bool = True,
) -> dict[str, int]:
    """Explicit canvases pass through (positive multiples of 32); the
    aspect-ratio resolver only applies when dimensions are omitted."""
    if height <= 0 or width <= 0 or height % 32 or width % 32:
        raise ValueError(f"H3 canvas must be positive multiples of 32, got {height}x{width}.")
    aligned_frames = align_num_frames(num_frames)
    latent_frames = video_latent_num_frames(aligned_frames)
    duration = aligned_frames / MINIMAX_H3_FPS
    if enforce_duration and not MINIMAX_H3_MIN_ALIGNED_FRAMES <= aligned_frames <= MINIMAX_H3_MAX_ALIGNED_FRAMES:
        raise ValueError(f"H3 generates {MINIMAX_H3_MIN_DURATION:g}-{MINIMAX_H3_MAX_DURATION:g} s at "
                         f"{MINIMAX_H3_FPS} fps; {aligned_frames} frames is {duration:.2f} s, outside the "
                         f"accepted {MINIMAX_H3_MIN_ALIGNED_FRAMES}-{MINIMAX_H3_MAX_ALIGNED_FRAMES} frame range.")
    return {
        "height": height,
        "width": width,
        "num_frames": aligned_frames,
        "latent_frame_count": latent_frames,
        "latent_height": height // 16,
        "latent_width": width // 16,
    }

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.prepare_resident

prepare_resident() -> None

Load the encoder, DiT, and both decoders once, before timed requests.

Source
def prepare_resident(self) -> None:
    """Load the encoder, DiT, and both decoders once, before timed requests."""
    if not self.resident or self._resident_components:
        return
    from fastvideo.mlx_runtime.minimax_h3_conditioner import ResidentNVFP4MiniMaxH3TextConditioner
    from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir
    import mlx.core as mx

    try:
        conditioner = self._load_conditioner()
        if not isinstance(conditioner, ResidentNVFP4MiniMaxH3TextConditioner):
            conditioner.close()
            raise ValueError("All-resident generation requires the packed NVFP4 text encoder.")
        self._resident_components["conditioner"] = conditioner
        self._resident_components["dit"] = _load_resident_dit(self.dit_checkpoint)
        self._resident_components["video_vae"] = self._load_video_vae(include_encoder=False)
        self._resident_components["audio_vae"] = mlx_h3_audio_vae_from_dir(self.model_root / "audio_vae",
                                                                           include_encoder=False)
        mx.eval(list(self._resident_components["audio_vae"].weights.values()))
        logger.info("H3 components resident: %.2f GiB active MLX memory", mx.get_active_memory() / 2**30)
    except Exception:
        self.close()
        raise

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.close

close() -> None
Source
def close(self) -> None:
    self._restore_quant_gemm()
    conditioner = self._resident_components.get("conditioner")
    if conditioner is not None:
        conditioner.close()
    self._resident_components.clear()
    _cleanup_mlx()
    previous = getattr(self, "_previous_wired_limit", None)
    if previous is not None:
        import mlx.core as mx

        _restore_metal_wired_limit(mx, previous)
        self._previous_wired_limit = None
    previous_memory = getattr(self, "_previous_memory_limit", None)
    if previous_memory is not None:
        import mlx.core as mx

        _restore_metal_memory_limit(mx, previous_memory)
        self._previous_memory_limit = None
    previous_cache = getattr(self, "_previous_cache_limit", None)
    if previous_cache is not None:
        import mlx.core as mx

        mx.set_cache_limit(previous_cache)
        self._previous_cache_limit = None

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.encode_prompt

encode_prompt(prompt: str) -> tuple[np.ndarray, np.ndarray]

Returns (hidden states (S, hidden), token tags). Uses the cache or the streamed conditioner.

Source
def encode_prompt(self, prompt: str) -> tuple[np.ndarray, np.ndarray]:
    """Returns (hidden states (S, hidden), token tags). Uses the cache or
    the streamed conditioner."""
    cache_key = self._prompt_cache_key(prompt)
    if cache_key is not None and cache_key.exists():
        data = np.load(cache_key)
        logger.info("Loaded prompt embeddings from cache %s", cache_key)
        return data["hidden_states"], data["token_tags"]

    if self.resident:
        self.prepare_resident()
        conditioner = self._resident_components["conditioner"]
    else:
        conditioner = self._load_conditioner()
    hidden, tags = conditioner.encode_prompt(prompt)
    if not self.resident:
        conditioner.close()
    _cleanup_mlx()
    if cache_key is not None:
        cache_key.parent.mkdir(parents=True, exist_ok=True)
        tmp_cache = cache_key.with_name(f".{cache_key.name}.tmp")
        try:
            with tmp_cache.open("wb") as handle:
                np.savez(handle, hidden_states=hidden, token_tags=tags)
            tmp_cache.replace(cache_key)
        finally:
            tmp_cache.unlink(missing_ok=True)
    return hidden, tags

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.load_prompt_cache

load_prompt_cache(path: str | Path) -> tuple[np.ndarray, np.ndarray]
Source
def load_prompt_cache(self, path: str | Path) -> tuple[np.ndarray, np.ndarray]:
    data = np.load(path)
    return data["hidden_states"], data["token_tags"]

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.prepare_references

prepare_references(references: Sequence[MiniMaxH3Reference], num_frames: int, *, max_pixels: int | None = None) -> list[MiniMaxH3PreparedReference]

Validate and decode the ordered reference list.

Decoding happens once, before anything is loaded, so a bad reference fails before the DiT and the encoders are resident.

Source
def prepare_references(
    self,
    references: Sequence[MiniMaxH3Reference],
    num_frames: int,
    *,
    max_pixels: int | None = None,
) -> list[MiniMaxH3PreparedReference]:
    """Validate and decode the ordered reference list.

    Decoding happens once, before anything is loaded, so a bad reference
    fails before the DiT and the encoders are resident.
    """
    validated = validate_references(list(references))
    target_sample_rate = self.audio_sample_rate()
    prepared = [
        prepare_reference(reference, num_frames, target_sample_rate, max_pixels=max_pixels)
        for reference in validated
    ]
    logger.info("Prepared %d reference(s): %s", len(prepared),
                ", ".join(reference.media_type for reference in prepared))
    return prepared

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.audio_sample_rate

audio_sample_rate() -> int

The audio VAE's sample rate, which reference soundtracks are resampled to.

Source
def audio_sample_rate(self) -> int:
    """The audio VAE's sample rate, which reference soundtracks are resampled to."""
    config_path = self.model_root / "audio_vae" / "config.json"
    if not config_path.is_file():
        return 32000
    return int(json.loads(config_path.read_text(encoding="utf-8")).get("sampling_rate", 32000))

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.encode_reference_prompt

encode_reference_prompt(prompt: str, references: list[MiniMaxH3PreparedReference]) -> tuple[np.ndarray, np.ndarray]

Encode the prompt with its ordered references as Qwen3-VL condition rows.

Source
def encode_reference_prompt(
    self,
    prompt: str,
    references: list[MiniMaxH3PreparedReference],
) -> tuple[np.ndarray, np.ndarray]:
    """Encode the prompt with its ordered references as Qwen3-VL condition rows."""
    from fastvideo.mlx_runtime.minimax_h3_ref_conditioning import encode_reference_prompt, load_vision_limits
    from fastvideo.mlx_runtime.minimax_h3_vision import Qwen3VLVisionConfig

    if self.resident:
        self.prepare_resident()
        conditioner = self._resident_components["conditioner"]
    else:
        conditioner = self._load_conditioner()
    try:
        config = Qwen3VLVisionConfig.from_config_json(self.conditioner_dir / "config.json")
        min_pixels, max_pixels = load_vision_limits(self.processor_dir())
        video_min_pixels, video_max_pixels = load_vision_limits(self.processor_dir(), video=True)
        hidden, tags = encode_reference_prompt(
            conditioner,
            prompt,
            references,
            vision_config=config,
            min_pixels=min_pixels,
            max_pixels=max_pixels,
            video_min_pixels=video_min_pixels,
            video_max_pixels=video_max_pixels,
        )
    finally:
        if not self.resident:
            conditioner.close()
    _cleanup_mlx()
    return hidden, tags

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.processor_dir

processor_dir() -> Path

Where the Qwen3-VL processor config lives, defaulting to the tokenizer dir.

Source
def processor_dir(self) -> Path:
    """Where the Qwen3-VL processor config lives, defaulting to the tokenizer dir."""
    for candidate in (self.model_root / "processor", self.tokenizer_dir):
        if (candidate / "preprocessor_config.json").is_file():
            return candidate
    return self.tokenizer_dir

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.has_conditioner_weights

has_conditioner_weights() -> bool
Source
def has_conditioner_weights(self) -> bool:
    marker = self.conditioner_dir / "model.safetensors.index.json"
    single = self.conditioner_dir / "model.safetensors"
    return marker.exists() or single.exists()

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.denoise

denoise(text_rows: np.ndarray, token_tags: np.ndarray, *, height: int, width: int, num_frames: int, audio_num_frames: int | None = None, video_temporal_scale: float = 1.0, seed: int, num_steps: int = 4, dit: Any | None = None, vsa_config: MiniMaxH3VSAConfig | None = None, inter_step_cooldown_s: float = 0.0, references: list[MiniMaxH3PreparedReference] | None = None, condition_video_rows: np.ndarray | None = None, condition_audio_rows: np.ndarray | None = None, reuse_prefix_kv: bool = False, prefix_kv_refresh_every: int = 0, prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB, prefix_kv_storage: str = 'bf16') -> tuple[np.ndarray, np.ndarray]

Denoise joint latents; returns (normalized target video rows, target audio rows).

With references the packed sequence is reference-prefixed and the encoder rows ride along in every forward, but only the target rows are stepped and returned: the references are conditioning, not output.

reuse_prefix_kv (Ref2VA only, lossy) records the text/reference rows' per-block K/V on the first step and runs later steps on the target rows alone; prefix_kv_refresh_every=R re-records every R steps (0 never). The cache is refused above prefix_kv_cache_max_gib.

Source
def denoise(
    self,
    text_rows: np.ndarray,
    token_tags: np.ndarray,
    *,
    height: int,
    width: int,
    num_frames: int,
    audio_num_frames: int | None = None,
    video_temporal_scale: float = 1.0,
    seed: int,
    num_steps: int = 4,
    dit: Any | None = None,
    vsa_config: MiniMaxH3VSAConfig | None = None,
    inter_step_cooldown_s: float = 0.0,
    references: list[MiniMaxH3PreparedReference] | None = None,
    condition_video_rows: np.ndarray | None = None,
    condition_audio_rows: np.ndarray | None = None,
    reuse_prefix_kv: bool = False,
    prefix_kv_refresh_every: int = 0,
    prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
    prefix_kv_storage: str = "bf16",
) -> tuple[np.ndarray, np.ndarray]:
    """Denoise joint latents; returns (normalized target video rows, target audio rows).

    With ``references`` the packed sequence is reference-prefixed and the
    encoder rows ride along in every forward, but only the target rows are
    stepped and returned: the references are conditioning, not output.

    ``reuse_prefix_kv`` (Ref2VA only, lossy) records the text/reference
    rows' per-block K/V on the first step and runs later steps on the
    target rows alone; ``prefix_kv_refresh_every=R`` re-records every R
    steps (0 never). The cache is refused above ``prefix_kv_cache_max_gib``.
    """
    import mlx.core as mx

    self.last_prefix_kv_stats: dict[str, object] = {"enabled": False}
    self.last_denoise_step_times = []
    if prefix_kv_refresh_every < 0:
        raise ValueError(f"prefix_kv_refresh_every must be >= 0, got {prefix_kv_refresh_every}.")
    if reuse_prefix_kv and references is None:
        raise ValueError("Prefix K/V reuse needs Ref2VA references (a reference-prefixed packed layout).")
    validate_prefix_kv_storage(prefix_kv_storage)

    geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=audio_num_frames is None)
    audio_frames = geometry["num_frames"] if audio_num_frames is None else align_num_frames(audio_num_frames)

    resident_dit = dit is None and self.resident
    if resident_dit:
        _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
        self.prepare_resident()
        dit = self._resident_components["dit"]
    owned_dit = dit is None
    if owned_dit:
        _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
        t0 = time.perf_counter()
        dit = load_mlx_h3_checkpoint(self.dit_checkpoint)
        logger.info("Loaded MLX H3 DiT from %s in %.1fs", self.dit_checkpoint, time.perf_counter() - t0)

    schedule = resolve_h3_denoise_schedule(self.model_root, num_steps)
    video_scheduler = schedule.video
    audio_scheduler = schedule.audio
    union = schedule.adaln_timesteps
    num_steps = schedule.num_steps
    # A FastH3 PDD export runs one fused head block per trained block; its
    # plan supplies the block partition, the per-modality node sigmas
    # (already on the schedulers) and the integration weights.
    pdd_plan = schedule.pdd_plan
    # The keyframe-noise timestep (0.999) is only exercised by FL2VA/Ref2VA
    # conditioning rows. A Ref2VA PDD ladder already carries it, so a
    # converted DiT's baked cache matches; this append is then a no-op.
    if references is not None:
        if video_temporal_scale != 1.0:
            # The reference-prefixed builder stretches only the target video
            # rows, which the CUDA packing never does either; refusing beats
            # silently ignoring the request.
            raise ValueError("Reference-conditioned packing has no temporal-scale stretch; "
                             "run Ref2VA without fast mode.")
        union = np.unique(np.concatenate([union, np.array([MINIMAX_H3_KEYFRAME_NOISE_AUG],
                                                          dtype=np.float32)])).astype(np.float32)
        if condition_video_rows is None or condition_audio_rows is None:
            condition_video_rows, condition_audio_rows = self._encode_reference_conditions(
                references, dit.patch_size, video_scheduler, seed)

    cache = getattr(dit, "_adaln_cache", None)
    if cache is None or not np.array_equal(cache.timesteps.astype(np.float32), union):
        if cache is not None:
            extra = np.setdiff1d(union, cache.timesteps)
            logger.info("Recomputing AdaLN cache for %d-step ladder (extra timesteps %s).", num_steps, extra)
        if _adaln_weights_dropped(dit):
            # An earlier request on this DiT already released the AdaLN projections.
            if not resident_dit:
                raise ValueError("This H3 DiT dropped its AdaLN weights for another step ladder; "
                                 "pass a freshly loaded DiT for a different num_steps.")
            logger.info("Reloading the resident H3 DiT for the %d-step ladder.", num_steps)
            cache = dit = None
            self._resident_components.pop("dit", None)
            _cleanup_mlx()
            dit = self._resident_components["dit"] = _load_resident_dit(self.dit_checkpoint)
        dit.precompute_adaln(union, drop_weights=True)

    if resident_dit:
        # The resident DiT keeps the previous request's VSA mode unless reset to the default.
        vsa_config = vsa_config or MiniMaxH3VSAConfig()
    if vsa_config is not None:
        dit.configure_vsa(vsa_config)
    if hasattr(dit, "reset_vsa_stats"):
        dit.reset_vsa_stats()

    if references is not None:
        layout = build_ref2va_packed_layout(
            np.asarray(token_tags, dtype=np.int64),
            references,
            geometry["latent_frame_count"],
            geometry["latent_height"],
            geometry["latent_width"],
            audio_latent_num_frames(audio_frames),
            patch_size=dit.patch_size,
        )
    else:
        layout = build_packed_layout(
            len(token_tags),
            geometry["latent_frame_count"],
            geometry["latent_height"],
            geometry["latent_width"],
            audio_latent_num_frames(audio_frames),
            patch_size=dit.patch_size,
            text_token_tags=np.asarray(token_tags, dtype=np.int64),
            video_temporal_scale=video_temporal_scale,
        )
    if getattr(dit, "vsa_config", None) is not None and dit.vsa_config.enabled:
        dit.prepare_vsa_geometry(layout)

    video_key, audio_key = mx.random.split(mx.random.key(seed))
    target_video_rows = int(layout.video_indices.shape[0] - layout.num_condition_video_rows)
    target_audio_rows = int(layout.audio_indices.shape[0] - layout.num_condition_audio_rows)
    x_v = mx.random.normal((target_video_rows, dit.patch_dim), key=video_key)
    x_a = mx.random.normal((target_audio_rows, dit.audio_in_channels), key=audio_key)
    if condition_video_rows is not None:
        if condition_video_rows.shape != (layout.num_condition_video_rows, dit.patch_dim):
            raise ValueError(f"Video condition rows {condition_video_rows.shape} do not match the layout's "
                             f"{layout.num_condition_video_rows} condition rows of width {dit.patch_dim}.")
        x_v = mx.concatenate([mx.array(condition_video_rows), x_v], axis=0)
    elif layout.num_condition_video_rows:
        raise ValueError("A reference-prefixed layout needs its encoded video condition rows.")
    if condition_audio_rows is not None:
        if condition_audio_rows.shape != (layout.num_condition_audio_rows, dit.audio_in_channels):
            raise ValueError(f"Audio condition rows {condition_audio_rows.shape} do not match the layout's "
                             f"{layout.num_condition_audio_rows} condition rows.")
        x_a = mx.concatenate([mx.array(condition_audio_rows), x_a], axis=0)
    elif layout.num_condition_audio_rows:
        raise ValueError("A reference-prefixed layout needs its encoded audio condition rows.")
    text = mx.array(text_rows.astype(np.float32))
    dit_forward_s = 0.0
    prefix_kv = None
    if reuse_prefix_kv:
        frozen_rows = frozen_prefix_rows(layout)
        estimated = estimate_prefix_kv_bytes(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                             prefix_kv_storage)
        check_prefix_kv_budget(estimated, prefix_kv_cache_max_gib, rows=frozen_rows)
        prefix_kv = MiniMaxH3PrefixKVCache(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                           prefix_kv_storage)
        logger.info("Prefix K/V reuse: %d frozen rows, ~%.2f GiB cache (%s), refresh_every=%d", frozen_rows,
                    estimated / 2**30, prefix_kv_storage, prefix_kv_refresh_every)
    recorded_steps: list[int] = []
    reused_steps: list[int] = []

    if pdd_plan is not None:
        # PDD targets start at the first node's noise level (0.999 * eps),
        # as the students were trained; conditioning rows stay untouched.
        x_v = mx.concatenate([
            x_v[:layout.num_condition_video_rows],
            x_v[layout.num_condition_video_rows:] * float(pdd_plan.node_sigmas["video"][0]),
        ],
                             axis=0)
        x_a = mx.concatenate([
            x_a[:layout.num_condition_audio_rows],
            x_a[layout.num_condition_audio_rows:] * float(pdd_plan.node_sigmas["audio"][0]),
        ],
                             axis=0)

    for step_index in range(num_steps):
        video_t = float(video_scheduler.timesteps[step_index])
        audio_t = float(audio_scheduler.timesteps[step_index])
        unique, inverse = build_row_timesteps(
            layout,
            video_timestep=video_t,
            audio_timestep=audio_t,
            condition_video_timestep=max(video_t, MINIMAX_H3_KEYFRAME_NOISE_AUG),
            condition_audio_timestep=1.0,
        )
        step_started = time.perf_counter()
        pdd_forward_kwargs = ({} if pdd_plan is None else {
            "pdd_block": pdd_plan.block(step_index),
            "pdd_integration_weights": pdd_plan.integration_weights,
        })
        record_step = prefix_kv is not None and (step_index == 0 or (prefix_kv_refresh_every > 0
                                                                     and step_index % prefix_kv_refresh_every == 0))
        if prefix_kv is not None and not record_step:
            video_velocity, audio_velocity = dit.forward_target_rows(
                x_v[layout.num_condition_video_rows:],
                x_a[layout.num_condition_audio_rows:],
                layout=layout,
                step_timesteps=unique,
                row_timestep_inverse=inverse,
                prefix_kv=prefix_kv,
                step_index=step_index,
                **pdd_forward_kwargs,
            )
            reused_steps.append(step_index)
        else:
            if record_step:
                pdd_forward_kwargs["record_prefix_kv"] = prefix_kv
                recorded_steps.append(step_index)
            video_velocity, audio_velocity = dit.forward_with_cache(
                x_v,
                x_a,
                text,
                layout=layout,
                step_timesteps=unique,
                row_timestep_inverse=inverse,
                step_index=step_index,
                **pdd_forward_kwargs,
            )
            video_velocity = video_velocity[layout.num_condition_video_rows:]
            audio_velocity = audio_velocity[layout.num_condition_audio_rows:]
        # Only the target rows are denoised: condition rows stay pinned at
        # the timestep the contract trained them at.
        if layout.num_condition_video_rows:
            x_v = mx.concatenate([
                x_v[:layout.num_condition_video_rows],
                video_scheduler.step(video_velocity, step_index, x_v[layout.num_condition_video_rows:])
            ],
                                 axis=0)
        else:
            x_v = video_scheduler.step(video_velocity, step_index, x_v)
        if layout.num_condition_audio_rows:
            x_a = mx.concatenate([
                x_a[:layout.num_condition_audio_rows],
                audio_scheduler.step(audio_velocity, step_index, x_a[layout.num_condition_audio_rows:])
            ],
                                 axis=0)
        else:
            x_a = audio_scheduler.step(audio_velocity, step_index, x_a)
        mx.eval(x_v, x_a)
        step_s = time.perf_counter() - step_started
        dit_forward_s += step_s
        self.last_denoise_step_times.append(step_s)
        logger.info("H3 denoise step %d/%d (sigma_video=%.4f, sigma_audio=%.4f) in %.2fs", step_index + 1,
                    num_steps, 1.0 - video_t, 1.0 - audio_t, step_s)
        del video_velocity, audio_velocity
        _cleanup_mlx()
        if inter_step_cooldown_s > 0 and step_index < num_steps - 1:
            logger.info("Thermal cooldown %.1fs before step %d...", inter_step_cooldown_s, step_index + 2)
            time.sleep(inter_step_cooldown_s)

    self.last_dit_forward_s = dit_forward_s
    self.last_prefix_kv_stats = {
        "enabled": prefix_kv is not None,
        "refresh_every": prefix_kv_refresh_every,
        "storage": prefix_kv_storage,
        "frozen_rows": 0 if prefix_kv is None else prefix_kv.rows,
        "cache_gib": 0.0 if prefix_kv is None else prefix_kv.nbytes / 2**30,
        "max_gib": prefix_kv_cache_max_gib,
        "recorded_steps": recorded_steps,
        "reused_steps": reused_steps,
    }
    prefix_kv = None
    stats = getattr(dit, "last_vsa_stats", None)
    self.last_vsa_stats = None if stats is None else {
        "enabled": bool(getattr(dit.vsa_config, "enabled", False)),
        "configured_sparsity": stats.configured_sparsity,
        "layer_sparsity": stats.layer_sparsity,
        "achieved_sparsity": stats.achieved_sparsity,
        "tile_size": stats.tile_size,
        "prefix_mode": stats.prefix_mode,
        "impl": stats.impl,
        "num_prefix_tiles": stats.num_prefix_tiles,
        "num_video_tiles": stats.num_video_tiles,
        "video_keep": stats.video_keep,
        "dense_fallback_reason": stats.dense_fallback_reason,
        "attention_calls": stats.attention_calls,
        "sparse_calls": stats.sparse_calls,
        "impl_counts": stats.impl_counts,
        "fallback_reasons": stats.fallback_reasons,
        "checkpoint_vsa_capable": bool(getattr(dit, "vsa_capable", False)),
    }

    video_rows = np.asarray(x_v, dtype=np.float32)[layout.num_condition_video_rows:]
    audio_rows = np.asarray(x_a, dtype=np.float32)[layout.num_condition_audio_rows:]
    if owned_dit:
        del dit
        _cleanup_mlx()
    return video_rows, audio_rows

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.decode_video

decode_video(video_rows: np.ndarray, *, height: int, width: int, num_frames: int, tiled: bool = True, vae_tile_height: int = 256, vae_tile_width: int = 256) -> np.ndarray

Normalized packed rows -> (T, H, W, 3) uint8 frames.

Source
def decode_video(self,
                 video_rows: np.ndarray,
                 *,
                 height: int,
                 width: int,
                 num_frames: int,
                 tiled: bool = True,
                 vae_tile_height: int = 256,
                 vae_tile_width: int = 256) -> np.ndarray:
    """Normalized packed rows -> (T, H, W, 3) uint8 frames."""
    _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
    import mlx.core as mx

    geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=False)
    if self.video_decode_backend == "taeh3":
        from fastvideo.mlx_runtime.minimax_h3_taeh3 import decode_latents_taeh3_mlx

        if self._dit_in_channels != 24:
            raise ValueError("TAEH3 requires a 24-channel H3 checkpoint.")
        latents = unpatchify_video_tokens(video_rows, geometry["latent_frame_count"], geometry["latent_height"],
                                          geometry["latent_width"], self._dit_in_channels, self._dit_patch_size)
        pixels = decode_latents_taeh3_mlx(latents,
                                          checkpoint_path=self.taeh3_checkpoint,
                                          dtype=self.vae_dtype,
                                          chunk_size=self.taeh3_chunk_size)
        frames = (pixels[0] * 255.0).astype(np.uint8)
        if frames.shape != (geometry["num_frames"], height, width, 3):
            raise RuntimeError(f"TAEH3 produced unexpected frame shape: {frames.shape}")
        _cleanup_mlx()
        return frames
    if self.resident:
        self.prepare_resident()
        vae = self._resident_components["video_vae"]
    else:
        vae = self._load_video_vae(include_encoder=False)
    expected_height = height // vae.spatial_compression_ratio
    expected_width = width // vae.spatial_compression_ratio
    if (geometry["latent_height"], geometry["latent_width"]) != (expected_height, expected_width):
        raise RuntimeError("H3 pipeline/VAE spatial compression mismatch: "
                           f"pipeline={(geometry['latent_height'], geometry['latent_width'])}, "
                           f"VAE={(expected_height, expected_width)}.")
    if vae.latent_channels != self._dit_in_channels:
        raise RuntimeError(
            f"H3 DiT/VAE latent-channel mismatch: DiT={self._dit_in_channels}, VAE={vae.latent_channels}.")
    latents = unpatchify_video_tokens(
        video_rows,
        geometry["latent_frame_count"],
        geometry["latent_height"],
        geometry["latent_width"],
        vae.latent_channels,
        self._dit_patch_size,
    )
    z = mx.array(latents)
    z = vae.denormalize_latents(z)
    decoded = vae.decode(z,
                         tiled=tiled,
                         tile_sample_min_height=min(geometry["height"], vae_tile_height),
                         tile_sample_min_width=min(geometry["width"], vae_tile_width))
    pixels = np.clip(np.asarray(vae.denormalize_pixels(decoded)), 0.0, 1.0)
    del vae, decoded, z
    _cleanup_mlx()
    frames = (pixels[0].transpose(1, 2, 3, 0) * 255.0).astype(np.uint8)  # (T, H, W, C)
    if frames.shape[0] != geometry["num_frames"]:
        raise RuntimeError(f"decoded {frames.shape[0]} frames, expected {geometry['num_frames']}")
    return frames

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.decode_audio

decode_audio(audio_rows: np.ndarray, *, num_frames: int) -> np.ndarray

Normalized packed audio rows -> stereo waveform (2, S) fp32 in [-1, 1].

Source
def decode_audio(self, audio_rows: np.ndarray, *, num_frames: int) -> np.ndarray:
    """Normalized packed audio rows -> stereo waveform (2, S) fp32 in [-1, 1]."""
    import mlx.core as mx

    from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir

    num_audio_latents = audio_latent_num_frames(align_num_frames(num_frames))
    latents = unpack_audio_tokens(audio_rows, num_audio_latents)
    if self.resident:
        self.prepare_resident()
        vae = self._resident_components["audio_vae"]
    else:
        vae = mlx_h3_audio_vae_from_dir(self.model_root / "audio_vae", include_encoder=False)
    z = vae.denormalize_latents(mx.array(latents))
    waveform = np.asarray(vae.decode(z))[:, 0, :]  # (B, 1, S) -> (B, S)
    del vae, z
    _cleanup_mlx()
    # Keep audio at least as long as the final video packet. Rounding down
    # by a fractional sample makes ffmpeg's ``-shortest`` drop frame 124.
    expected_samples = _audio_sample_count(align_num_frames(num_frames))
    if waveform.shape[-1] < expected_samples:
        waveform = np.pad(waveform, ((0, 0), (0, expected_samples - waveform.shape[-1])))
    return np.clip(waveform[:, :expected_samples], -1.0, 1.0)

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.mux

mux(frames: np.ndarray, waveform: np.ndarray, output_path: str | Path, fps: int = MINIMAX_H3_FPS, sample_rate: int = 32000) -> Path

H.264 video + AAC stereo audio, A/V durations within one frame.

Source
def mux(self,
        frames: np.ndarray,
        waveform: np.ndarray,
        output_path: str | Path,
        fps: int = MINIMAX_H3_FPS,
        sample_rate: int = 32000) -> Path:
    """H.264 video + AAC stereo audio, A/V durations within one frame."""
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_video = output_path.with_suffix(".tmp.mp4")
    tmp_audio = output_path.with_suffix(".tmp.wav")

    pcm = (np.clip(waveform.T, -1.0, 1.0) * 32767.0).astype("<i2")  # (S, 2)
    import wave

    try:
        with wave.open(str(tmp_audio), "wb") as handle:
            handle.setnchannels(2)
            handle.setsampwidth(2)
            handle.setframerate(sample_rate)
            handle.writeframes(pcm.tobytes())

        height, width = frames.shape[1:3]
        ffmpeg = shutil.which("ffmpeg")
        if ffmpeg is None:
            raise RuntimeError("ffmpeg is required for MP4 muxing.")
        subprocess.run(
            [
                ffmpeg,
                "-y",
                "-loglevel",
                "error",
                "-f",
                "rawvideo",
                "-pix_fmt",
                "rgb24",
                "-s",
                f"{width}x{height}",
                "-r",
                str(fps),
                "-i",
                "-",
                "-i",
                str(tmp_audio),
                "-c:v",
                "libx264",
                "-preset",
                "medium",
                "-crf",
                "18",
                "-pix_fmt",
                "yuv420p",
                "-c:a",
                "aac",
                "-b:a",
                "192k",
                "-shortest",
                "-movflags",
                "+faststart",
                str(tmp_video),
            ],
            input=frames.tobytes(),
            check=True,
        )
        tmp_video.replace(output_path)
    finally:
        tmp_audio.unlink(missing_ok=True)
        tmp_video.unlink(missing_ok=True)
    return output_path

fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.generate

generate(prompt: str, *, output_path: str | Path, references: Sequence[MiniMaxH3Reference] | None = None, height: int = 480, width: int = 832, num_frames: int = 124, seed: int = 0, num_steps: int = 4, save_frames: bool = False, tiled_video_decode: bool = True, vae_tile_height: int = 256, vae_tile_width: int = 256, fast: bool = False, fast_factor: int = 2, fast_sharpen: float = 0.6, rife_weights_dir: str | Path | None = None, fast_spatial: bool = False, fast_spatial_scale: int = 2, fast_spatial_upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE, fast_spatial_sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN, vsa: bool | None = None, vsa_sparsity: float | None = None, vsa_tile_size: int | None = None, vsa_ref_policy: str | None = None, vsa_ref_keep_rate: float | None = None, vsa_prefix_mode: str = 'exempt', vsa_dense_first_n_steps: int = 0, vsa_dense_layers: tuple[int, ...] = (), vsa_impl: str = 'auto', inter_step_cooldown_s: float = 0.0, reference_max_pixels: int | None = None, reuse_prefix_kv: bool = False, prefix_kv_refresh_every: int = 0, prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB, prefix_kv_storage: str = 'bf16') -> GenerationResult
Source
def generate(
    self,
    prompt: str,
    *,
    output_path: str | Path,
    references: Sequence[MiniMaxH3Reference] | None = None,
    height: int = 480,
    width: int = 832,
    num_frames: int = 124,
    seed: int = 0,
    num_steps: int = 4,
    save_frames: bool = False,
    tiled_video_decode: bool = True,
    vae_tile_height: int = 256,
    vae_tile_width: int = 256,
    fast: bool = False,
    fast_factor: int = 2,
    fast_sharpen: float = 0.6,
    rife_weights_dir: str | Path | None = None,
    fast_spatial: bool = False,
    fast_spatial_scale: int = 2,
    fast_spatial_upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE,
    fast_spatial_sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN,
    vsa: bool | None = None,
    vsa_sparsity: float | None = None,
    vsa_tile_size: int | None = None,
    vsa_ref_policy: str | None = None,
    vsa_ref_keep_rate: float | None = None,
    vsa_prefix_mode: str = "exempt",
    vsa_dense_first_n_steps: int = 0,
    vsa_dense_layers: tuple[int, ...] = (),
    vsa_impl: str = "auto",
    inter_step_cooldown_s: float = 0.0,
    reference_max_pixels: int | None = None,
    reuse_prefix_kv: bool = False,
    prefix_kv_refresh_every: int = 0,
    prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
    prefix_kv_storage: str = "bf16",
) -> GenerationResult:
    timings: dict[str, float] = {}
    peaks: dict[str, float] = {}

    if fast_sharpen < 0:
        raise ValueError(f"fast_sharpen must be non-negative, got {fast_sharpen}.")
    if inter_step_cooldown_s < 0:
        raise ValueError(f"inter_step_cooldown_s must be non-negative, got {inter_step_cooldown_s}.")
    _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
    _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
    # Unset VSA knobs come from the snapshot's FastH3 contract; ``vsa=None``
    # turns VSA on for a VSA contract and ``vsa=False`` opts out.
    vsa_settings = resolve_vsa_settings(
        load_fasth3_inference_contract(self.model_root),
        vsa=vsa,
        sparsity=vsa_sparsity,
        tile_size=vsa_tile_size,
        ref_policy=vsa_ref_policy,
        ref_keep_rate=vsa_ref_keep_rate,
        checkpoint_capable=mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
    )
    vsa = vsa_settings.enabled
    vsa_config = MiniMaxH3VSAConfig(
        enabled=vsa,
        sparsity=vsa_settings.sparsity,
        tile_size=vsa_settings.tile_size,
        ref_policy=vsa_settings.ref_policy,
        ref_keep_rate=vsa_settings.ref_keep_rate,
        prefix_mode=vsa_prefix_mode,  # type: ignore[arg-type]
        dense_first_n_steps=vsa_dense_first_n_steps,
        dense_layers=vsa_dense_layers,
        impl=vsa_impl,  # type: ignore[arg-type]
    ) if vsa else MiniMaxH3VSAConfig()
    if vsa_config.enabled and not mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint):
        raise dense_only_vsa_error(self.dit_checkpoint)
    spatial_plan = plan_fast_spatial(
        height,
        width,
        scale=fast_spatial_scale,
        upsample_mode=fast_spatial_upsample_mode,
        sharpen=fast_spatial_sharpen,
    ) if fast_spatial else None
    _preflight_media_dependencies(
        fast=fast,
        fast_sharpen=fast_sharpen,
        rife_weights_dir=rife_weights_dir,
        fast_spatial=fast_spatial,
    )
    if spatial_plan is not None:
        canvas_height, canvas_width = spatial_plan.canvas_height, spatial_plan.canvas_width
    else:
        canvas_height, canvas_width = _model_canvas_size(height, width)
    target_geometry = self.resolve_geometry(canvas_height, canvas_width, num_frames)
    fast_plan = plan_fast_temporal(target_geometry["num_frames"], fast_factor) if fast else None
    video_num_frames = fast_plan.source_frames if fast_plan is not None else target_geometry["num_frames"]
    video_temporal_scale = fast_plan.video_temporal_scale if fast_plan is not None else 1.0
    video_geometry = self.resolve_geometry(
        canvas_height,
        canvas_width,
        video_num_frames,
        enforce_duration=fast_plan is None,
    )
    logger.info(
        "Geometry: output=%dx%dx%d model=%dx%dx%d audio_frames=%d fast=%s fast_spatial=%s",
        width,
        height,
        target_geometry["num_frames"],
        canvas_width,
        canvas_height,
        video_geometry["num_frames"],
        target_geometry["num_frames"],
        fast_plan,
        spatial_plan,
    )

    if self.video_decode_backend == "taeh3":
        from fastvideo.mlx_runtime.minimax_h3_taeh3 import ensure_taeh3_checkpoint

        started = time.perf_counter()
        ensure_taeh3_checkpoint(self.taeh3_checkpoint)
        timings["decoder_prepare_s"] = time.perf_counter() - started
        logger.warning("TAEH3 is an approximate preview decoder; reconstruction differs from the full H3 VAE.")

    prepared_references: list[MiniMaxH3PreparedReference] | None = None
    if references is not None:
        prepared_references = self.prepare_references(references,
                                                      video_geometry["num_frames"],
                                                      max_pixels=reference_max_pixels)

    _reset_peak_memory()
    started = time.perf_counter()
    if prepared_references is None:
        text_rows, token_tags = self.encode_prompt(prompt)
    else:
        text_rows, token_tags = self.encode_reference_prompt(prompt, prepared_references)
    timings["condition_s"] = time.perf_counter() - started
    peaks["condition_gib"] = _peak_memory_gib()
    _cleanup_mlx()

    _reset_peak_memory()
    started = time.perf_counter()
    video_rows, audio_rows = self.denoise(
        text_rows,
        token_tags,
        height=video_geometry["height"],
        width=video_geometry["width"],
        num_frames=video_geometry["num_frames"],
        audio_num_frames=target_geometry["num_frames"] if fast_plan is not None else None,
        video_temporal_scale=video_temporal_scale,
        seed=seed,
        num_steps=num_steps,
        vsa_config=vsa_config,
        inter_step_cooldown_s=inter_step_cooldown_s,
        references=prepared_references,
        reuse_prefix_kv=reuse_prefix_kv,
        prefix_kv_refresh_every=prefix_kv_refresh_every,
        prefix_kv_cache_max_gib=prefix_kv_cache_max_gib,
        prefix_kv_storage=prefix_kv_storage,
    )
    timings["denoise_s"] = time.perf_counter() - started
    timings["dit_forward_s"] = float(getattr(self, "last_dit_forward_s", 0.0))
    peaks["denoise_gib"] = _peak_memory_gib()
    del text_rows
    _cleanup_mlx()

    _reset_peak_memory()
    started = time.perf_counter()
    frames = self.decode_video(
        video_rows,
        height=video_geometry["height"],
        width=video_geometry["width"],
        num_frames=video_geometry["num_frames"],
        tiled=tiled_video_decode,
        vae_tile_height=vae_tile_height,
        vae_tile_width=vae_tile_width,
    )
    if spatial_plan is not None:
        frames = _center_crop_frames(frames, spatial_plan.stage1_height, spatial_plan.stage1_width)
    else:
        frames = _center_crop_frames(frames, height, width)
    timings["video_decode_s"] = time.perf_counter() - started
    peaks["video_decode_gib"] = _peak_memory_gib()
    _cleanup_mlx()

    if fast_plan is not None:
        from fastvideo.mlx_runtime.rife_interp import interpolate_to_frame_count, load_model

        _reset_peak_memory()
        started = time.perf_counter()
        model = load_model(weights_dir=str(rife_weights_dir) if rife_weights_dir is not None else None)
        try:
            interpolated = interpolate_to_frame_count(
                frames,
                target_geometry["num_frames"],
                model=model,
            )
            if spatial_plan is None:
                interpolated = _sharpen_frames(interpolated, fast_sharpen)
            frames = np.stack(interpolated)
            if frames.shape[0] != target_geometry["num_frames"]:
                raise RuntimeError(
                    f"RIFE produced {frames.shape[0]} frames, expected {target_geometry['num_frames']}.")
            del interpolated
            timings["rife_s"] = time.perf_counter() - started
            peaks["rife_gib"] = _peak_memory_gib()
        finally:
            load_model.cache_clear()
            del model
            _cleanup_mlx()

    if spatial_plan is not None:
        started = time.perf_counter()
        # One sharpen pass, at full resolution: RIFE and the resample soften
        # for the same reason, so the stronger requested amount is applied
        # once instead of stacking two unsharp masks.
        sharpen = spatial_plan.sharpen if fast_plan is None else max(spatial_plan.sharpen, fast_sharpen)
        frames = np.stack(
            upsample_frames(
                frames,
                width=spatial_plan.target_width,
                height=spatial_plan.target_height,
                mode=spatial_plan.upsample_mode,
                sharpen=sharpen,
            ))
        timings["spatial_upsample_s"] = time.perf_counter() - started

    _reset_peak_memory()
    started = time.perf_counter()
    waveform = self.decode_audio(audio_rows, num_frames=target_geometry["num_frames"])
    timings["audio_decode_s"] = time.perf_counter() - started
    peaks["audio_decode_gib"] = _peak_memory_gib()
    _cleanup_mlx()

    started = time.perf_counter()
    video_path = self.mux(frames, waveform, output_path)
    timings["mux_s"] = time.perf_counter() - started
    timings["generate_s"] = sum(
        timings.get(key, 0.0) for key in ("decoder_prepare_s", "condition_s", "denoise_s", "video_decode_s",
                                          "rife_s", "spatial_upsample_s", "audio_decode_s", "mux_s"))

    result = GenerationResult(
        video_path=str(video_path),
        frames=frames if save_frames else None,
        waveform=waveform,
        sample_rate=32000,
        timings=timings,
        peak_memory_gib=peaks,
        video_decode_backend=self.video_decode_backend,
        prefix_kv={
            **getattr(self, "last_prefix_kv_stats", {"enabled": False}),
            "step_times_s":
            list(getattr(self, "last_denoise_step_times", [])),
        },
        vsa=getattr(self, "last_vsa_stats", None) or {
            "enabled": vsa_config.enabled,
            "checkpoint_vsa_capable": mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
        },
    )
    logger.info("Generation complete: %s | timings=%s peaks=%s", video_path, timings, peaks)
    return result