[Bugfix][MiniMax-H3] Apply pre-commit formatting

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>

docs/.nav.yml

@@ -37,7 +37,7 @@ nav:
       - Image-To-Video: user_guide/examples/online_serving/image_to_video.md
       - Speech-To-Video: user_guide/examples/online_serving/speech_to_video.md
       - Text-To-Image: user_guide/examples/online_serving/text_to_image.md
-      - Text-To-Speech: user_guide/examples/online_serving/text_to_speech.md
+      - Text-To-Speech (Online Serving): user_guide/examples/online_serving/text_to_speech.md
       - Text-To-Video: user_guide/examples/online_serving/text_to_video.md
   - General:
     - usage/*
@@ -127,6 +127,8 @@ nav:
         - Async Omni Output Materialization: design/feature/omni_async_output_materialization.md
         - Automatic Prefix Caching: design/feature/prefix_caching.md
         - Realtime AR-Diffusion: design/feature/realtime_ar_diffusion.md
+      - Model-Specific Capabilities:
+        - MiniMax H3 Temporal VAE Chunks: design/feature/minimax_h3_temporal_vae_chunks.md
       - Communication:
         - OmniConnector Implementations:
           - Mooncake Store Connector: design/feature/omni_connectors/mooncake_store_connector.md

docs/design/architecture_overview.md

@@ -170,6 +170,10 @@ because the user-visible output is diffusion media rather than a token stream.
 The
 [MiniMax-H3 recipe](https://github.com/vllm-project/vllm-omni/blob/main/recipes/MiniMaxAI/MiniMax-H3.md)
 contains the hardware-specific profiles and warmup requirements.
+The internal, opt-in
+[temporal VAE chunk callback](feature/minimax_h3_temporal_vae_chunks.md)
+documents the narrower released-source contract for publishing committed video
+frames during decode. It does not change the complete-output serving topology.
 
 ### Cosmos3: unified MoT reasoner and generator
 

docs/design/feature/minimax_h3_temporal_vae_chunks.md

@@ -0,0 +1,260 @@
+# MiniMax H3 temporal VAE chunk callback
+
+!!! warning "Experimental internal capability"
+
+    This feature is an opt-in model integration seam. The current MiniMax H3
+    pipeline still uses complete-output VAE decode, and no CLI option, serving
+    path, transport, or MP4 sink enables temporal chunks by default.
+
+## Status and scope
+
+MiniMax H3 decodes video latents in temporal windows, but the released VAE
+normally exposes only the complete reconstructed video. The temporal callback
+publishes frames after the released decoder has completed overlap blending and
+padding removal. This gives a future runtime consumer a stable point at which
+it can schedule representation conversion, D2H transfer, and encoding while
+the VAE continues decoding later windows.
+
+The implementation remains model-local:
+
+- `vae.py` owns VAE topology and the thin
+  `decode_latent_with_chunks()` facade;
+- `chunked_decode.py` owns the normalized callback contract, metadata
+  validation, and fixed VAE-group rendezvous points; and
+- `temporal_chunks.py` owns the source-gated compatibility planner and the
+  released temporal assembly control flow.
+
+Crop policy, typed media, transport, backpressure, encoding, and public output
+remain runtime responsibilities. The model code does not schedule requests or
+select a presentation format.
+
+## Execution flow
+
+The default and opt-in paths are intentionally separate:
+
+```text
+Default:
+latent -> MiniMaxH3VideoVAE.decode_latent() -> complete RGB video
+
+Opt-in producer seam:
+latent
+  -> MiniMaxH3VideoVAE.decode_latent_with_chunks()
+  -> group-local preflight and callback-owner validation
+  -> released temporal decode / overlap blending / padding removal
+  -> borrowed decoder-domain chunk
+  -> normalize + clone into owned RGB float32 BCTHW [0, 1]
+  -> synchronous callback
+  -> complete RGB video is still returned
+```
+
+Calling `decode_latent_with_chunks()` is the opt-in. Supplying no callback on a
+VAE peer does not select the default decoder; every rank in a VAE patch group
+must enter the chunked method, and only the output rank receives the callback.
+
+## Producer contract
+
+The normalized callback has this structural signature:
+
+```python
+def callback(
+    frames: torch.Tensor,
+    /,
+    *,
+    chunk_index: int,
+    total_chunks: int,
+    frame_start: int,
+    is_final: bool,
+) -> None:
+    ...
+```
+
+For a successful decode:
+
+- `frames` is owned, contiguous RGB float32 in `BCTHW` layout and `[0, 1]`;
+- callback storage does not alias the returned complete video;
+- the consumer may retain or mutate a chunk;
+- `chunk_index` is zero-based, contiguous, and strictly increasing;
+- `total_chunks` is positive and stable for the decode;
+- `frame_start` equals the end of the previous published range;
+- chunks are nonempty and contain only temporally committed frames;
+- exactly one chunk has `is_final=True`; and
+- concatenating chunks along time exactly reconstructs the returned video.
+
+The callback runs synchronously on the producer's current device stream. A
+consumer that reads the tensor on another stream must establish ordering, for
+example with an event, and retain the allocation for that stream's lifetime.
+A slow callback also blocks VAE production. An overlap-oriented runtime
+consumer should therefore acquire bounded capacity, enqueue side-stream work,
+and return promptly instead of encoding frames inside the callback.
+
+Chunks retain the decoded VAE canvas dimensions. A future runtime integration
+must apply the same requested-height and requested-width crop as the complete
+MiniMax H3 pipeline before declaring the media transport-ready.
+
+The raw compatibility callback is deliberately narrower. It receives borrowed
+decoder-domain tensors and must not retain or mutate them. The coordinator is
+the ownership boundary that normalizes and clones raw chunks before invoking
+the public callback.
+
+## Temporal planning and terminal publication
+
+The released configuration advances five latent tokens per temporal window and
+includes a two-token overlap, so each `_adaptive_decode` input normally contains
+seven tokens. After temporal scaling, pre-padding removal, and five-frame
+overlap blending, the normal committed unit is 17 output frames. The final
+remainder depends on the requested frame plan.
+
+Representative released frame plans include:
+
+| Output frames | Published chunks |
+| ---: | --- |
+| 124 | `7 x 17 + 5` |
+| 209 | `12 x 17 + 5` |
+| 243 | `14 x 17 + 5` |
+| 362 | `21 x 17 + 5` |
+
+The terminal callback is held until the complete frame plan passes its logical
+frame, padding, written-frame, chunk-count, and finality checks. Earlier chunks
+are semantically stable, but they do not imply that the overall request has
+succeeded; a consumer must discard or abort partial output if a later decoder
+failure terminates the request.
+
+## Source and configuration gate
+
+The released VAE has no public temporal-callback API. vLLM-Omni therefore owns
+a compatibility implementation for one exact released code contract. The
+reference revision identifies its provenance, but runtime acceptance is based
+on source content, configuration, and structure rather than repository or
+weight metadata:
+
+- reference Hugging Face revision, not independently checked at runtime:
+  `42ed227ee7df40d41602854ae760620d6eb651fe`;
+- sorted 15-file top-level `video_vae/*.py` manifest (excluding
+  `__init__.py`) SHA-256:
+  `30e64afbaa940696bf8bfe2c86003a4b1666c84b22fa3aa12403a3e6a03f705d`;
+- clip length `17`;
+- token chunk, overlap, and drop `5`, `2`, and `3`;
+- temporal ratio `4`;
+- frame pre-padding and overlap `3` and `5`; and
+- no isolated first or last frame.
+
+The adapter does not verify Hugging Face revision metadata, repository identity,
+or model weights. The manifest pin detects Python compatibility drift; it is
+not a security boundary for remote-code execution. The source manifest, listed
+temporal attributes, structural contract, plan, and dtype are validated before
+the first `_adaptive_decode` call. A mismatch disables only the opt-in chunk path.
+`decode_latent()` remains available and performs no source fingerprint or chunk
+rendezvous.
+
+## Supported models
+
+Support in this table means the internal producer contract above, not public
+serving or end-to-end transport support.
+
+| Model or component | Source contract | Status | Validation |
+| --- | --- | --- | --- |
+| MiniMax H3 FL2VA video VAE | Released manifest/listed temporal attributes above (reference `42ed227e`) | Supported, opt-in | Same-environment full-output parity and L20X/B300 feasibility measured on the implementation through PR commit `60e9bbad` |
+| MiniMax H3 Ref2VA video VAE | Same released manifest/listed temporal attributes | Source-compatible; not separately benchmarked | Ref2VA and FL2VA use byte-identical video-VAE source and configuration in the reference snapshot |
+| MiniMax H3 with a different Python manifest or listed temporal attribute | Different enforced contract | Unsupported for chunk mode | Fails closed before decoder collectives; complete decode remains available |
+| Same manifest/temporal attributes with different weights or other component metadata | Not checked by this gate | Not independently qualified | The adapter does not inspect weights, repository identity, revision metadata, or arbitrary component configuration |
+| Other video models | Any | Unsupported | No generic temporal VAE producer ABI is declared by this feature |
+
+## Feature compatibility
+
+| Dimension or combination | Status | Constraint |
+| --- | --- | --- |
+| Existing complete-output decode | Supported and default | No source check, callback object, or callback rendezvous |
+| Explicit chunk decode | Supported internally | Caller invokes `decode_latent_with_chunks()` directly |
+| VAE patch parallel size 1 | Supported by the model seam | Caller supplies the callback only on the request output owner; no WORLD rendezvous is added |
+| VAE patch parallel size equal to the full DiT group | Supported when data parallel size is 1 | Every VAE-group rank calls; exactly group rank 0 owns the callback |
+| Partial VAE patch subgroup | Unsupported | MiniMax H3 currently accepts size 1 or the full DiT group only |
+| Data parallel size greater than 1 with VAE patch parallel size 1 | Model seam only | Per-replica callback ownership is possible, but no serving/runtime consumer is wired |
+| Data parallel size greater than 1 with VAE patch parallel size greater than 1 | Unsupported | Fails before decoder collectives because no replica-local VAE patch group is defined |
+| Too few spatial tiles for the VAE group | Correctness fallback | Decodes rank-locally and restores native tiling state; slower than patch parallel decode |
+| Device memory and full-output assembly | Not reduced or bounded | The compatibility path still assembles and returns the full decode while allocating owned callback chunks; no peak-HBM reduction is claimed |
+| DiT cache, attention, TP, SP, HSDP, LoRA, or quantization features | No direct callback dependency | This seam runs after denoising and does not independently qualify combinations beyond the MiniMax H3 model's own matrix |
+| Layerwise or distributed component offload | Not separately validated | Component residency must place the video VAE before decode |
+| Step execution or request batching | Not integrated | No pipeline caller binds a runtime chunk consumer |
+| Requested-size crop and device uint8 preparation | Not implemented here | Runtime-owned follow-up |
+| Pinned D2H, SHM/ZMQ, or cross-process transport | Not implemented here | Runtime-owned follow-up |
+| Complete MP4 or fragmented MP4 sink | Not implemented here | Existing serving behavior still waits for complete output |
+| Audio VAE chunks and incremental A/V composition | Not supported | Video chunks do not imply audio or request finality |
+| Cancellation and multi-output lease ownership | Not implemented here | Requires a bounded runtime lifecycle |
+| NVIDIA CUDA | Feasibility evidence | L20X and B300 measurements apply to the implementation through `60e9bbad`; the current coordination refactor was not reprofiled |
+| AMD ROCm | Not validated for this feature | Base MiniMax H3 support is documented separately and does not imply temporal-chunk validation |
+| Ascend, Intel, or other accelerators | No feature support claim | Consult the model support table independently; this design adds no platform qualification |
+
+## Distributed and failure semantics
+
+When VAE patch parallelism is active, all VAE-group ranks run the same
+preflight before entering checkpoint collectives. Source/planner errors and the
+resolved temporal concat dtype are reconciled at fixed group rendezvous points.
+
+Callback ownership is also a group decision. Rank `r` contributes `r + 1` when
+it owns a callback and zero otherwise. A group SUM equals one only when rank 0
+is the sole owner; every invalid owner configuration therefore fails
+identically on all ranks before decode.
+
+Callback exceptions are retained on the output rank. Further publication stops,
+but every rank completes the same remaining native decoder collective schedule.
+The original exception is then raised on the owner and
+`MiniMaxH3ChunkCallbackPeerError` on peers.
+
+Exceptions raised inside the checkpoint's native distributed decode cannot in
+general be reconciled by the adapter: a peer may already be blocked inside a
+checkpoint-owned collective. Resolving such failures requires native collective
+handling or supervisor-level timeout and worker-group teardown. The process
+group must not be assumed reusable after such a failure.
+
+## Validation evidence
+
+The contract suite covers ordinary and padded frame plans, blend boundaries,
+owned storage and mutation isolation, source-manifest/listed-temporal-attribute
+drift, callback failure, and healthy subsequent requests. Real Gloo processes cover
+rank-asymmetric preflight failures, dtype disagreement, two-rank callback
+failure propagation, three-rank wrong-owner rejection, insufficient-tile
+fallback, and group recovery for failures that occur at adapter-owned
+rendezvous points.
+
+Four-device L20X and independent four- and eight-device B300 experiments on the
+implementation through PR commit `60e9bbad` showed exact serial-versus-chunked
+frame/MP4 parity within each environment and lower VAE-to-complete-MP4 latency
+when an experimental consumer was attached. Hashes are not claimed to match
+across different hardware/software environments. The later coordination and
+module-boundary refactors were revalidated with CPU and real Gloo tests but were
+not reprofiled on GPUs. These are feasibility measurements, not a promise for
+default serving, because this change does not contain the transport or encoder
+consumer. See [RFC #6872](https://github.com/vllm-project/vllm-omni/issues/6872)
+and [PR #6885](https://github.com/vllm-project/vllm-omni/pull/6885) for the
+measurement provenance and limitations.
+
+## Generalization boundary and follow-ups
+
+This feature does not define a repository-wide `SupportsVideoChunkDecode`
+interface. H3 temporal blending, Wan causal caches, LTX temporal/spatial halos,
+and pipeline-level generation chunks have different commit and failure
+boundaries. A generic producer contract should be extracted only after a second
+model proves the same semantic event without `Any` arguments or model-name
+dispatch.
+
+The reusable future boundary is the emitted, committed video-media chunk, not a
+universal latent-to-VAE method. It should extend the typed media work in
+[RFC #6541](https://github.com/vllm-project/vllm-omni/issues/6541) and
+[PR #6615](https://github.com/vllm-project/vllm-omni/pull/6615), while keeping
+model-specific halo, cache, padding, blending, distributed, and source-gating
+logic inside each model adapter.
+
+Runtime follow-ups own:
+
+- typed video-plus-audio media and request/output identity;
+- requested-size crop and device preparation;
+- bounded leases, backpressure, cancellation, and cleanup;
+- D2H/IPC transport and observability; and
+- complete-MP4 and fragmented-MP4 consumers.
+
+## Related documentation
+
+- [Diffusion model integration](../module/diffusion/diffusion_model_integration.md)
+- [VAE patch parallelism](vae_parallel.md)
+- [Async diffusion output](async_diffusion_output.md)
+- [MiniMax H3 recipe](https://github.com/vllm-project/vllm-omni/blob/main/recipes/MiniMaxAI/MiniMax-H3.md)

docs/design/index.md

@@ -25,6 +25,10 @@ implementation contract; it is not, by itself, a general support claim.
 - [Automatic Prefix Caching in Omni Models](feature/prefix_caching.md)
 - [Realtime AR-Diffusion Sessions](feature/realtime_ar_diffusion.md)
 
+### Model-specific capabilities
+
+- [MiniMax H3 Temporal VAE Chunks](feature/minimax_h3_temporal_vae_chunks.md)
+
 ### Communication
 
 #### OmniConnector implementations

docs/design/module/diffusion/diffusion_model_integration.md

@@ -57,3 +57,10 @@ cross-stage routing.
 
 Test registry selection, checkpoint loading, minimal inference, input and output
 contracts, and every declared optional capability.
+
+## Model-specific capability documents
+
+- [MiniMax H3 temporal VAE chunk callback](../../feature/minimax_h3_temporal_vae_chunks.md)
+  defines the source-gated producer seam and its boundary with future runtime
+  media transport. It is an internal opt-in capability, not a public serving
+  feature.

recipes/MiniMaxAI/MiniMax-H3.md

@@ -1142,3 +1142,4 @@ vllm serve "${MODEL_ROOT}/FL2VA" \
 - [Supported models](../../docs/models/supported_models.md)
 - [Video API](../../docs/serving/videos_api.md)
 - [Diffusion parallelism](../../docs/user_guide/diffusion/parallelism/overview.md)
+- [Temporal VAE chunk callback design](../../docs/design/feature/minimax_h3_temporal_vae_chunks.md)

vllm_omni/diffusion/models/minimax_h3/chunked_decode.py

@@ -0,0 +1,391 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Model-local coordination for MiniMax H3 temporal chunk decoding.
+
+This boundary owns released-code compatibility and reconciles preflight and
+callback failures at fixed VAE-group rendezvous points. Cropping,
+representation conversion, transport, and encoding remain runtime
+responsibilities.
+"""
+
+from __future__ import annotations
+
+from contextlib import AbstractContextManager
+from typing import Protocol
+
+import torch
+import torch.distributed as dist
+import torch.nn as nn
+
+from .temporal_chunks import (
+    MiniMaxH3TemporalChunkCompatibilityError,
+    MiniMaxH3TemporalDecodePlan,
+    _TemporalChunkCallbackSignature,
+    decode_minimax_h3_temporal_chunks_compat,
+    prepare_minimax_h3_temporal_chunks_compat,
+)
+
+
+class MiniMaxH3VideoChunkCallback(_TemporalChunkCallbackSignature, Protocol):
+    """Consume one ordered, temporally committed H3 video chunk.
+
+    ``frames`` is an owned, contiguous float32 RGB tensor in ``BCTHW`` layout
+    and the ``[0, 1]`` value range. The callback runs synchronously on the
+    producer's current device stream. Before reading it on another stream, the
+    consumer must establish ordering from the producer stream (for example via
+    an event) and retain the allocation for that stream's lifetime. The
+    consumer may retain or mutate the tensor; it does not alias the returned
+    complete decode.
+    """
+
+
+class MiniMaxH3ChunkedDecodeUnsupportedError(RuntimeError):
+    """Raised when loaded MiniMax H3 code cannot honor the chunk contract."""
+
+
+class MiniMaxH3ChunkCallbackPeerError(RuntimeError):
+    """Raised on peer VAE ranks when the output-rank callback fails."""
+
+
+class _MiniMaxH3ChunkDecodeHost(Protocol):
+    """Narrow VAE capabilities required by the chunk coordinator."""
+
+    model: nn.Module
+
+    def _denormalize_latent(self, latent: torch.Tensor) -> torch.Tensor: ...
+
+    def _decode_tiling_context(
+        self,
+        latent: torch.Tensor,
+    ) -> AbstractContextManager: ...
+
+    def _normalize_decoded_frames(
+        self,
+        decoded: torch.Tensor,
+    ) -> torch.Tensor: ...
+
+
+class _MiniMaxH3ChunkDecodeCoordinator:
+    """Run source-gated chunk decode with group-local failure semantics.
+
+    The coordinator is a plain Python object. It caches only successful source
+    validation; request callbacks, tensors, metadata, and process groups remain
+    local to :meth:`decode`.
+
+    Errors raised inside the checkpoint decoder after native collectives begin
+    cannot be reconciled here: another rank may already be blocked inside that
+    collective. Resolving such a failure requires native collective handling or
+    supervisor-level timeout and worker-group teardown; the process group cannot
+    be assumed reusable.
+    """
+
+    def __init__(self) -> None:
+        self._sources_validated = False
+
+    @staticmethod
+    def _build_execution_signature(
+        latent: torch.Tensor,
+        plan: MiniMaxH3TemporalDecodePlan,
+    ) -> tuple[int, ...]:
+        """Describe every shape and plan field used by the temporal schedule."""
+
+        dtype_codes: dict[torch.dtype | None, int] = {
+            None: 0,
+            torch.float16: 1,
+            torch.bfloat16: 2,
+            torch.float32: 3,
+            torch.float64: 4,
+        }
+        return (
+            # Keep the layout versioned so adding a field cannot silently make
+            # an older and newer worker appear compatible.
+            1,
+            *[int(size) for size in latent.shape],
+            dtype_codes.get(latent.dtype, -1),
+            *[int(size) for size in plan.latent.shape],
+            dtype_codes.get(plan.latent.dtype, -1),
+            int(plan.num_chunks),
+            int(plan.total_frames),
+            int(plan.pad_frames),
+            int(plan.output_frames),
+            len(plan.chunk_frames),
+            *[int(frame_count) for frame_count in plan.chunk_frames],
+            dtype_codes.get(plan.temporal_dtype, -1),
+        )
+
+    @staticmethod
+    def _is_output_rank(
+        chunk_callback: MiniMaxH3VideoChunkCallback | None,
+        *,
+        group: dist.ProcessGroup | None,
+        device: torch.device,
+    ) -> bool:
+        if group is None:
+            return chunk_callback is not None
+
+        group_rank = dist.get_rank(group)
+        # A valid configuration has one contribution: rank 0 contributes 1.
+        # A non-zero owner contributes at least 2, and multiple owners sum to
+        # at least 3, so every rank makes the same decision from one reduction.
+        callback_owner = torch.tensor(
+            [group_rank + 1 if chunk_callback is not None else 0],
+            dtype=torch.int64,
+            device=device,
+        )
+        dist.all_reduce(callback_owner, op=dist.ReduceOp.SUM, group=group)
+        if int(callback_owner.item()) != 1:
+            raise ValueError("MiniMax-H3 chunk decode requires a callback on exactly VAE group rank 0")
+        return group_rank == 0
+
+    @staticmethod
+    def _synchronize_callback_error(
+        callback_error: BaseException | None,
+        *,
+        group: dist.ProcessGroup | None,
+        device: torch.device,
+    ) -> None:
+        if group is None:
+            if callback_error is not None:
+                raise callback_error
+            return
+
+        is_output_rank = dist.get_rank(group) == 0
+        failed = torch.tensor(
+            [int(is_output_rank and callback_error is not None)],
+            dtype=torch.int32,
+            device=device,
+        )
+        dist.broadcast(
+            failed,
+            src=dist.get_global_rank(group, 0),
+            group=group,
+        )
+        if int(failed.item()) == 0:
+            return
+        if is_output_rank:
+            assert callback_error is not None
+            raise callback_error
+        raise MiniMaxH3ChunkCallbackPeerError("MiniMax-H3 video chunk callback failed on this VAE group's output rank")
+
+    @staticmethod
+    def _synchronize_preflight_error(
+        preflight_error: Exception | None,
+        temporal_dtype: torch.dtype | None,
+        execution_signature: tuple[int, ...] | None,
+        *,
+        group: dist.ProcessGroup | None,
+        device: torch.device,
+    ) -> None:
+        if group is None:
+            if preflight_error is not None:
+                raise preflight_error
+            return
+        failed = torch.tensor(
+            [int(preflight_error is not None)],
+            dtype=torch.int32,
+            device=device,
+        )
+        dist.all_reduce(failed, op=dist.ReduceOp.MAX, group=group)
+        if int(failed.item()) == 0:
+            dtype_codes = {
+                None: 0,
+                torch.float16: 1,
+                torch.bfloat16: 2,
+                torch.float32: 3,
+            }
+            code = torch.tensor(
+                [dtype_codes.get(temporal_dtype, -1)],
+                dtype=torch.int32,
+                device=device,
+            )
+            lowest = code.clone()
+            highest = code.clone()
+            dist.all_reduce(lowest, op=dist.ReduceOp.MIN, group=group)
+            dist.all_reduce(highest, op=dist.ReduceOp.MAX, group=group)
+            if int(lowest.item()) == int(highest.item()) and int(lowest.item()) >= 0:
+                if execution_signature is None:
+                    raise RuntimeError("MiniMax-H3 temporal decode preflight produced no execution signature")
+
+                # The plan length is shape-dependent, so first agree on a
+                # common tensor size and then compare the complete signatures.
+                # This keeps all ranks in tensor collectives and catches a
+                # schedule mismatch before entering the decoder's collectives.
+                signature_length = torch.tensor(
+                    [len(execution_signature)],
+                    dtype=torch.int64,
+                    device=device,
+                )
+                max_signature_length = signature_length.clone()
+                dist.all_reduce(
+                    max_signature_length,
+                    op=dist.ReduceOp.MAX,
+                    group=group,
+                )
+                padded_signature = torch.full(
+                    (int(max_signature_length.item()),),
+                    -2,
+                    dtype=torch.int64,
+                    device=device,
+                )
+                padded_signature[: len(execution_signature)] = torch.tensor(
+                    execution_signature,
+                    dtype=torch.int64,
+                    device=device,
+                )
+                gathered_signatures = [torch.empty_like(padded_signature) for _ in range(dist.get_world_size(group))]
+                dist.all_gather(gathered_signatures, padded_signature, group=group)
+                if any(not torch.equal(padded_signature, peer) for peer in gathered_signatures):
+                    raise MiniMaxH3ChunkedDecodeUnsupportedError(
+                        "MiniMax-H3 VAE ranks built incompatible temporal decode plans"
+                    )
+                return
+            raise MiniMaxH3ChunkedDecodeUnsupportedError(
+                "MiniMax-H3 VAE ranks selected incompatible temporal concat dtypes"
+            )
+        if preflight_error is not None:
+            raise preflight_error
+        raise MiniMaxH3ChunkedDecodeUnsupportedError("A peer VAE rank rejected the MiniMax-H3 temporal chunk preflight")
+
+    def decode(
+        self,
+        host: _MiniMaxH3ChunkDecodeHost,
+        latent: torch.Tensor,
+        chunk_callback: MiniMaxH3VideoChunkCallback | None,
+        *,
+        group: dist.ProcessGroup | None,
+    ) -> torch.Tensor:
+        """Decode one latent while publishing committed normalized chunks."""
+
+        plan: MiniMaxH3TemporalDecodePlan | None = None
+        preflight_error: Exception | None = None
+        if latent.ndim != 5:
+            preflight_error = MiniMaxH3ChunkedDecodeUnsupportedError(
+                f"MiniMax-H3 video latent must be rank 5, got {tuple(latent.shape)}"
+            )
+        else:
+            try:
+                denormalized = host._denormalize_latent(latent)
+                plan = prepare_minimax_h3_temporal_chunks_compat(
+                    host.model,
+                    denormalized,
+                    validate_sources=not self._sources_validated,
+                )
+                self._sources_validated = True
+                del denormalized
+            except MiniMaxH3TemporalChunkCompatibilityError as exc:
+                preflight_error = MiniMaxH3ChunkedDecodeUnsupportedError(str(exc))
+            except Exception as exc:  # noqa: BLE001
+                # Every VAE rank must reach the preflight reduction even when
+                # one node's private planner or environment fails locally.
+                exc.__traceback__ = None
+                preflight_error = exc
+        self._synchronize_preflight_error(
+            preflight_error,
+            plan.temporal_dtype if plan is not None else None,
+            self._build_execution_signature(latent, plan) if plan is not None else None,
+            group=group,
+            device=latent.device,
+        )
+        assert plan is not None
+        is_output_rank = self._is_output_rank(
+            chunk_callback,
+            group=group,
+            device=latent.device,
+        )
+        callback_error: BaseException | None = None
+        next_chunk_index = 0
+        expected_total_chunks: int | None = None
+        next_frame_start = 0
+        saw_final = False
+
+        def publish(
+            decoded_chunk: torch.Tensor,
+            *,
+            chunk_index: int,
+            total_chunks: int,
+            frame_start: int,
+            is_final: bool,
+        ) -> None:
+            nonlocal callback_error, expected_total_chunks
+            nonlocal next_chunk_index, next_frame_start, saw_final
+            if not is_output_rank or callback_error is not None:
+                return
+            try:
+                if expected_total_chunks is None:
+                    expected_total_chunks = total_chunks
+                if (
+                    saw_final
+                    or total_chunks <= 0
+                    or total_chunks != expected_total_chunks
+                    or chunk_index != next_chunk_index
+                    or chunk_index >= total_chunks
+                    or frame_start != next_frame_start
+                    or is_final != (chunk_index + 1 == total_chunks)
+                ):
+                    raise RuntimeError(
+                        "MiniMax-H3 temporal backend emitted discontinuous "
+                        "chunk metadata: "
+                        f"index={chunk_index}/{next_chunk_index}, "
+                        f"total={total_chunks}/{expected_total_chunks}, "
+                        f"frame_start={frame_start}/{next_frame_start}, "
+                        f"final={is_final}, after_final={saw_final}"
+                    )
+                frames = host._normalize_decoded_frames(decoded_chunk)
+                frames = frames.clone(memory_format=torch.contiguous_format)
+                frame_count = int(frames.shape[2])
+                if frame_count <= 0:
+                    raise RuntimeError("MiniMax-H3 temporal backend emitted an empty chunk")
+                next_chunk_index += 1
+                next_frame_start += frame_count
+                saw_final = is_final
+                assert chunk_callback is not None
+                chunk_callback(
+                    frames,
+                    chunk_index=chunk_index,
+                    total_chunks=total_chunks,
+                    frame_start=frame_start,
+                    is_final=is_final,
+                )
+            except BaseException as exc:  # noqa: BLE001
+                # Keep a failed GPU chunk and its callback locals out of the
+                # traceback while peer ranks finish the remaining collectives.
+                exc.__traceback__ = None
+                callback_error = exc
+
+        # Do not catch and reduce exceptions from this region. A peer may
+        # already be blocked in one of the checkpoint's native tile
+        # collectives, in which case a new adapter-level collective would also
+        # deadlock. A native backend or supervisor must handle timeout and
+        # worker-group teardown.
+        with host._decode_tiling_context(latent):
+            decoded = decode_minimax_h3_temporal_chunks_compat(
+                host.model,
+                plan,
+                publish,
+            )
+        frames = host._normalize_decoded_frames(decoded)
+        if (
+            is_output_rank
+            and callback_error is None
+            and (not saw_final or next_chunk_index != expected_total_chunks or next_frame_start != int(frames.shape[2]))
+        ):
+            callback_error = RuntimeError(
+                "MiniMax-H3 temporal backend did not reconstruct the full "
+                "decode: "
+                f"final={saw_final}, "
+                f"chunks={next_chunk_index}/{expected_total_chunks}, "
+                f"frames={next_frame_start}/{frames.shape[2]}"
+            )
+        self._synchronize_callback_error(
+            callback_error,
+            group=group,
+            device=frames.device,
+        )
+        return frames
+
+
+__all__ = [
+    "MiniMaxH3ChunkCallbackPeerError",
+    "MiniMaxH3ChunkedDecodeUnsupportedError",
+    "MiniMaxH3VideoChunkCallback",
+]

vllm_omni/diffusion/models/minimax_h3/temporal_chunks.py

@@ -0,0 +1,414 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Source-gated temporal chunk compatibility for the released MiniMax H3 VAE."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib
+import inspect
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol
+
+import torch
+import torch.nn as nn
+
+MINIMAX_H3_RELEASED_REVISION = "42ed227ee7df40d41602854ae760620d6eb651fe"
+# Compatibility pins detect upstream code drift; they are not a trust or
+# security boundary for remote code execution.
+MINIMAX_H3_RELEASED_BUNDLE_SHA256 = "30e64afbaa940696bf8bfe2c86003a4b1666c84b22fa3aa12403a3e6a03f705d"
+
+_RELEASED_TEMPORAL_CONFIG = {
+    "clip_length": 17,
+    "tokens_chunk_size": 5,
+    "token_overlap": 2,
+    "token_drop": 3,
+    "vae_ratio_t": 4,
+    "frame_pre_padding": 3,
+    "frame_overlap": 5,
+    "isolated_first_frame": False,
+    "isolated_last_frame": False,
+}
+
+
+class MiniMaxH3TemporalChunkCompatibilityError(RuntimeError):
+    """Raised when remote H3 code cannot use the local compatibility path."""
+
+
+class _TemporalChunkCallbackSignature(Protocol):
+    """Shared callable shape; each producer defines tensor semantics."""
+
+    def __call__(
+        self,
+        frames: torch.Tensor,
+        /,
+        *,
+        chunk_index: int,
+        total_chunks: int,
+        frame_start: int,
+        is_final: bool,
+    ) -> None: ...
+
+
+@dataclass(slots=True)
+class MiniMaxH3TemporalDecodePlan:
+    latent: torch.Tensor
+    num_chunks: int
+    total_frames: int
+    pad_frames: int
+    output_frames: int
+    chunk_frames: tuple[int, ...]
+    temporal_dtype: torch.dtype | None
+
+
+class _RawChunkAssembly:
+    def __init__(
+        self,
+        *,
+        callback: _TemporalChunkCallbackSignature,
+        plan: MiniMaxH3TemporalDecodePlan,
+    ) -> None:
+        self.callback = callback
+        self.plan = plan
+        self.frames: torch.Tensor | None = None
+        self.write_pos = 0
+        self.logical_frames = 0
+        self.dropped_frames = 0
+        self.chunk_index = 0
+        self.pending_final: tuple[torch.Tensor, int, int] | None = None
+        self.contract_error: RuntimeError | None = None
+
+    def write(self, part: torch.Tensor) -> None:
+        part_frames = int(part.shape[2])
+        if part_frames <= 0:
+            return
+        self.logical_frames += part_frames
+        remaining = self.plan.output_frames - self.write_pos
+        copy_frames = min(part_frames, max(0, remaining))
+        if copy_frames <= 0:
+            self.dropped_frames += part_frames
+            return
+        if self.chunk_index >= len(self.plan.chunk_frames):
+            self.contract_error = RuntimeError("MiniMax-H3 compatibility decode emitted an extra chunk")
+        else:
+            expected_frames = self.plan.chunk_frames[self.chunk_index]
+            if copy_frames != expected_frames and self.contract_error is None:
+                self.contract_error = RuntimeError(
+                    "MiniMax-H3 compatibility chunk size mismatch: "
+                    f"chunk={self.chunk_index}, frames={copy_frames}/{expected_frames}"
+                )
+
+        emitted = part[:, :, :copy_frames]
+        if self.frames is None:
+            output_shape = list(emitted.shape)
+            output_shape[2] = self.plan.output_frames
+            self.frames = torch.empty(
+                output_shape,
+                dtype=emitted.dtype,
+                device=emitted.device,
+            )
+        frame_start = self.write_pos
+        frame_stop = frame_start + copy_frames
+        self.frames[:, :, frame_start:frame_stop].copy_(emitted)
+        if frame_stop == self.plan.output_frames:
+            self.pending_final = (emitted, self.chunk_index, frame_start)
+        elif self.contract_error is None:
+            self.callback(
+                emitted,
+                chunk_index=self.chunk_index,
+                total_chunks=len(self.plan.chunk_frames),
+                frame_start=frame_start,
+                is_final=False,
+            )
+        self.chunk_index += 1
+        self.write_pos = frame_stop
+        self.dropped_frames += part_frames - copy_frames
+
+    def finish(self) -> torch.Tensor:
+        if self.contract_error is not None:
+            raise self.contract_error
+        if self.frames is None or self.pending_final is None:
+            raise RuntimeError("MiniMax-H3 compatibility decode produced no terminal chunk")
+        if (
+            self.logical_frames != self.plan.total_frames
+            or self.dropped_frames != self.plan.pad_frames
+            or self.write_pos != self.plan.output_frames
+            or self.chunk_index != len(self.plan.chunk_frames)
+        ):
+            raise RuntimeError(
+                "MiniMax-H3 compatibility frame plan mismatch: "
+                f"logical={self.logical_frames}/{self.plan.total_frames}, "
+                f"dropped={self.dropped_frames}/{self.plan.pad_frames}, "
+                f"written={self.write_pos}/{self.plan.output_frames}, "
+                f"chunks={self.chunk_index}/{len(self.plan.chunk_frames)}"
+            )
+
+        final_frames, final_index, final_start = self.pending_final
+        self.callback(
+            final_frames,
+            chunk_index=final_index,
+            total_chunks=len(self.plan.chunk_frames),
+            frame_start=final_start,
+            is_final=True,
+        )
+        return self.frames
+
+
+def _source_bundle_sha256(model: nn.Module) -> tuple[str | None, str | None]:
+    try:
+        source = inspect.getsourcefile(type(model))
+    except TypeError:
+        return None, None
+    if source is None:
+        return None, None
+    package_dir = Path(source).parent
+    files = sorted(path for path in package_dir.glob("*.py") if path.name != "__init__.py")
+    digest = hashlib.sha256()
+    try:
+        for path in files:
+            digest.update(path.name.encode())
+            digest.update(b"\0")
+            digest.update(path.read_bytes())
+            digest.update(b"\0")
+    except OSError:
+        return str(package_dir), None
+    return str(package_dir), digest.hexdigest()
+
+
+def _validate_released_sources(model: nn.Module) -> None:
+    source_dir, actual = _source_bundle_sha256(model)
+    if actual != MINIMAX_H3_RELEASED_BUNDLE_SHA256:
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            "The vLLM-owned MiniMax-H3 chunk adapter supports only remote code "
+            f"from MiniMaxAI/MiniMax-H3@{MINIMAX_H3_RELEASED_REVISION}: "
+            f"bundle={actual!r} ({source_dir}). "
+            "The normal complete-output decode remains available."
+        )
+
+
+def _validate_released_config(model: nn.Module) -> None:
+    actual = {name: getattr(model, name, None) for name in _RELEASED_TEMPORAL_CONFIG}
+    if actual != _RELEASED_TEMPORAL_CONFIG:
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            "The vLLM-owned MiniMax-H3 chunk adapter requires the released temporal "
+            f"configuration {_RELEASED_TEMPORAL_CONFIG}, found {actual}. "
+            "The normal complete-output decode remains available."
+        )
+
+
+def _validate_compat_contract(model: nn.Module, latent: torch.Tensor) -> None:
+    if latent.ndim != 5 or int(latent.shape[2]) <= 0:
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            f"MiniMax-H3 video latent must have positive rank-5 time, got {tuple(latent.shape)}"
+        )
+    if not bool(getattr(model, "use_3d_conv", False)):
+        raise MiniMaxH3TemporalChunkCompatibilityError("MiniMax-H3 temporal chunk decode requires the 3D decoder")
+    if bool(getattr(model, "training", False)):
+        raise MiniMaxH3TemporalChunkCompatibilityError("MiniMax-H3 temporal chunk decode is inference-only")
+    required_methods = (
+        "_adaptive_decode",
+        "_decode_temporal_output_frame_plan",
+        "_decode_temporal_pad_frames",
+        "blend",
+    )
+    missing = [name for name in required_methods if not callable(getattr(model, name, None))]
+    if missing:
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            "Loaded MiniMax-H3 VAE lacks required released-code methods: " + ", ".join(missing)
+        )
+
+
+def _truncate_chunk_frames(
+    logical_chunk_frames: list[int],
+    output_frames: int,
+) -> tuple[int, ...]:
+    remaining = output_frames
+    result: list[int] = []
+    for logical_frames in logical_chunk_frames:
+        emitted_frames = min(logical_frames, max(0, remaining))
+        if emitted_frames > 0:
+            result.append(emitted_frames)
+            remaining -= emitted_frames
+    if remaining != 0:
+        raise RuntimeError(f"MiniMax-H3 compatibility chunk plan cannot cover output: remaining={remaining}")
+    return tuple(result)
+
+
+def _build_decode_plan(
+    model: nn.Module,
+    latent: torch.Tensor,
+    temporal_dtype: torch.dtype | None,
+) -> MiniMaxH3TemporalDecodePlan:
+    pseudo_total_tokens = int(latent.shape[2]) + int(model.token_drop)  # type: ignore[attr-defined]
+    tokens_chunk_size = int(model.tokens_chunk_size)  # type: ignore[attr-defined]
+    remainder = pseudo_total_tokens % tokens_chunk_size
+    pad_tokens = 0 if remainder == 0 else tokens_chunk_size - remainder
+    pseudo_total_tokens += pad_tokens
+    num_chunks = pseudo_total_tokens // tokens_chunk_size - int(model.token_drop > 0)  # type: ignore[attr-defined]
+    if num_chunks <= 0:
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            f"MiniMax-H3 temporal chunk plan is empty for latent T={latent.shape[2]}"
+        )
+
+    temporal_latent = latent
+    if pad_tokens:
+        pad = temporal_latent[:, :, -1:].repeat(1, 1, pad_tokens, 1, 1)
+        temporal_latent = torch.cat((temporal_latent, pad), dim=2)
+    total_frames, pad_frames, output_frames = model._decode_temporal_output_frame_plan(  # type: ignore[attr-defined]
+        temporal_latent,
+        None,
+        None,
+        num_chunks,
+        pad_tokens,
+    )
+    if output_frames <= 0:
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            f"MiniMax-H3 temporal chunk plan has {output_frames} output frames"
+        )
+
+    chunk_decoded_frames = tokens_chunk_size * int(model.vae_ratio_t)  # type: ignore[attr-defined]
+    split_count = int(model.token_drop > 0) + 1  # type: ignore[attr-defined]
+    logical_chunk_frames: list[int] = []
+    final_overlap_frames = 0
+    for temporal_index in range(num_chunks):
+        token_start = temporal_index * tokens_chunk_size
+        token_stop = token_start + tokens_chunk_size + int(model.token_overlap)  # type: ignore[attr-defined]
+        clip_tokens = max(
+            0,
+            min(token_stop, int(temporal_latent.shape[2])) - min(token_start, int(temporal_latent.shape[2])),
+        )
+        clip_frames = clip_tokens * int(model.vae_ratio_t)  # type: ignore[attr-defined]
+        for split_index in range(split_count):
+            frame_start = split_index * chunk_decoded_frames
+            frame_stop = min(frame_start + chunk_decoded_frames, clip_frames)
+            part_frames = max(
+                0,
+                frame_stop - frame_start - int(model.frame_pre_padding),  # type: ignore[attr-defined]
+            )
+            if split_index == 0:
+                if part_frames > 0:
+                    logical_chunk_frames.append(part_frames)
+            else:
+                final_overlap_frames = part_frames
+    if final_overlap_frames > 0:
+        logical_chunk_frames.append(final_overlap_frames)
+    if sum(logical_chunk_frames) != int(total_frames):
+        raise RuntimeError(
+            "MiniMax-H3 compatibility planner disagrees with released remote code: "
+            f"frames={sum(logical_chunk_frames)}/{total_frames}"
+        )
+    chunk_frames = _truncate_chunk_frames(logical_chunk_frames, int(output_frames))
+    return MiniMaxH3TemporalDecodePlan(
+        latent=temporal_latent,
+        num_chunks=num_chunks,
+        total_frames=int(total_frames),
+        pad_frames=int(pad_frames),
+        output_frames=int(output_frames),
+        chunk_frames=chunk_frames,
+        temporal_dtype=temporal_dtype,
+    )
+
+
+def resolve_minimax_h3_temporal_cat_dtype(model: nn.Module) -> torch.dtype | None:
+    module = importlib.import_module(type(model).__module__)
+    resolver = getattr(module, "_resolve_temporal_cat_dtype", None)
+    if resolver is None:
+        return None
+    if not callable(resolver):
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            "Loaded MiniMax-H3 VAE has a non-callable temporal dtype resolver"
+        )
+    try:
+        dtype = resolver()
+    except ValueError as exc:
+        raise MiniMaxH3TemporalChunkCompatibilityError(str(exc)) from None
+    if dtype is not None and not isinstance(dtype, torch.dtype):
+        raise MiniMaxH3TemporalChunkCompatibilityError(
+            f"Loaded MiniMax-H3 VAE returned invalid temporal dtype {dtype!r}"
+        )
+    return dtype
+
+
+def decode_minimax_h3_temporal_chunks_compat(
+    model: nn.Module,
+    plan: MiniMaxH3TemporalDecodePlan,
+    callback: _TemporalChunkCallbackSignature,
+) -> torch.Tensor:
+    """Mirror the released Apache-2.0 H3 temporal assembler with callbacks.
+
+    Keep this implementation and the source/config fingerprints above in sync
+    with ``MiniMaxAI/MiniMax-H3@42ed227e``. Unknown remote code fails closed
+    before this function is called.
+
+    Callback tensors are borrowed decoder-domain views and must not be retained
+    or mutated. The high-level adapter owns normalization and storage isolation.
+    """
+
+    chunk_decoded_frames = int(model.tokens_chunk_size) * int(model.vae_ratio_t)  # type: ignore[attr-defined]
+    split_count = int(model.token_drop > 0) + 1  # type: ignore[attr-defined]
+    assembly = _RawChunkAssembly(callback=callback, plan=plan)
+    overlap: torch.Tensor | None = None
+
+    for temporal_index in range(plan.num_chunks):
+        token_start = temporal_index * int(model.tokens_chunk_size)  # type: ignore[attr-defined]
+        token_stop = token_start + int(model.tokens_chunk_size) + int(model.token_overlap)  # type: ignore[attr-defined]
+        chunk_latent = plan.latent[:, :, token_start:token_stop]
+        decoded = model._adaptive_decode(chunk_latent)  # type: ignore[attr-defined]
+        if plan.temporal_dtype is not None and decoded.dtype != plan.temporal_dtype:
+            decoded = decoded.to(plan.temporal_dtype)
+        if decoded.device != plan.latent.device:
+            decoded = decoded.to(plan.latent.device)
+
+        for split_index in range(split_count):
+            frame_start = split_index * chunk_decoded_frames
+            frame_stop = min(frame_start + chunk_decoded_frames, int(decoded.shape[2]))
+            part = decoded[:, :, frame_start:frame_stop]
+            part = part[:, :, int(model.frame_pre_padding) :]  # type: ignore[attr-defined]
+            if split_index == 0:
+                if overlap is not None:
+                    part = model.blend(  # type: ignore[attr-defined]
+                        overlap,
+                        part,
+                        int(model.frame_overlap),  # type: ignore[attr-defined]
+                        dim=-3,
+                    )
+                    overlap = None
+                assembly.write(part)
+            else:
+                overlap = part.contiguous()
+            del part
+
+        if temporal_index == plan.num_chunks - 1 and overlap is not None:
+            assembly.write(overlap)
+            overlap = None
+        del decoded, chunk_latent
+
+    return assembly.finish()
+
+
+def prepare_minimax_h3_temporal_chunks_compat(
+    model: nn.Module,
+    latent: torch.Tensor,
+    *,
+    validate_sources: bool = True,
+) -> MiniMaxH3TemporalDecodePlan:
+    """Build a validated plan without entering decoder collectives."""
+
+    if validate_sources:
+        _validate_released_sources(model)
+    _validate_released_config(model)
+    _validate_compat_contract(model, latent)
+    temporal_dtype = resolve_minimax_h3_temporal_cat_dtype(model)
+    return _build_decode_plan(model, latent, temporal_dtype)
+
+
+__all__ = [
+    "MINIMAX_H3_RELEASED_REVISION",
+    "MINIMAX_H3_RELEASED_BUNDLE_SHA256",
+    "MiniMaxH3TemporalDecodePlan",
+    "MiniMaxH3TemporalChunkCompatibilityError",
+    "decode_minimax_h3_temporal_chunks_compat",
+    "prepare_minimax_h3_temporal_chunks_compat",
+    "resolve_minimax_h3_temporal_cat_dtype",
+]

vllm_omni/diffusion/models/minimax_h3/vae.py

@@ -21,12 +21,22 @@ from vllm.logger import init_logger
 from vllm_omni.diffusion.distributed.autoencoders.distributed_vae_executor import (
     DistributedVaeMixin,
 )
-from vllm_omni.diffusion.distributed.parallel_state import get_world_group
+from vllm_omni.diffusion.distributed.parallel_state import (
+    get_data_parallel_world_size,
+    get_world_group,
+    model_parallel_is_initialized,
+)
 from vllm_omni.diffusion.offloader.module_residency import (
     BoundedAllocatorCache,
     PinnedModuleStager,
 )
 
+from .chunked_decode import (
+    MiniMaxH3ChunkCallbackPeerError,
+    MiniMaxH3ChunkedDecodeUnsupportedError,
+    MiniMaxH3VideoChunkCallback,
+    _MiniMaxH3ChunkDecodeCoordinator,
+)
 from .ops import install_h3_vae_optimizations
 from .packed_tokens import minimax_h3_patchify_video_latent
 
@@ -153,6 +163,7 @@ class MiniMaxH3VideoVAE(nn.Module, DistributedVaeMixin):
         self.use_slicing = False
         self.parallel_size = 1
         self.device_module = torch.get_device_module()
+        self._chunk_decode_coordinator: _MiniMaxH3ChunkDecodeCoordinator | None = None
 
     def load_to_device(self) -> None:
         if self._stager is not None:
@@ -265,6 +276,59 @@ class MiniMaxH3VideoVAE(nn.Module, DistributedVaeMixin):
     def is_distributed_enabled(self) -> bool:
         return self.parallel_size > 1 and dist.is_initialized()
 
+    def _denormalize_latent(self, latent: torch.Tensor) -> torch.Tensor:
+        channels = int(self.config_dict["latent_channels"])
+        mean = torch.tensor(
+            self.config_dict["latents_mean"],
+            device=latent.device,
+            dtype=latent.dtype,
+        ).view(1, channels, 1, 1, 1)
+        std = torch.tensor(
+            self.config_dict["latents_std"],
+            device=latent.device,
+            dtype=latent.dtype,
+        ).view(1, channels, 1, 1, 1)
+        return latent * std + mean
+
+    def _decode_tiling_context(self, latent: torch.Tensor) -> AbstractContextManager:
+        # The checkpoint hands rank r the tiles ``range(r, num_tiles, sp_size)``
+        # and then rejects an empty share inside the gather. A rank with no
+        # tiles raises and leaves the collective while the others block in it
+        # forever, so too few tiles hangs the whole stage rather than failing
+        # it. Tile count depends only on the latent shape, so every rank takes
+        # this branch together.
+        num_tiles = self._decoder_tile_count(latent)
+        if self.parallel_size > 1 and num_tiles < self.parallel_size:
+            logger.warning_once(
+                "MiniMax-H3 VAE decode splits into %d tile(s) but the tile group has "
+                "%d ranks; decoding rank-locally for this shape instead, which is "
+                "slower but avoids ranks without tiles hanging the collective.",
+                num_tiles,
+                self.parallel_size,
+            )
+            return self._rank_local_tiling()
+        return nullcontext()
+
+    def _normalize_decoded_frames(self, decoded: torch.Tensor) -> torch.Tensor:
+        frames = self.model.processor.revert_tensor(decoded)
+        if frames.ndim == 4:
+            frames = frames.unsqueeze(0).transpose(1, 2)
+        if frames.ndim != 5:
+            raise ValueError(f"unexpected decoded video shape {tuple(frames.shape)}")
+        return frames.float()
+
+    def _chunk_process_group(self) -> dist.ProcessGroup | None:
+        if not self.is_distributed_enabled():
+            return None
+        if model_parallel_is_initialized() and get_data_parallel_world_size() > 1:
+            raise MiniMaxH3ChunkedDecodeUnsupportedError(
+                "MiniMax-H3 temporal chunks do not yet support combining data parallelism with VAE patch parallelism"
+            )
+        group = self._native_parallel_state().get("sp_process_group")
+        if group is None or dist.get_world_size(group) != self.parallel_size:
+            raise RuntimeError("MiniMax-H3 VAE chunk decode has an invalid spatial-parallel group")
+        return group
+
     @torch.inference_mode()
     def encode_image(self, image: Image.Image) -> torch.Tensor:
         previous_parallel = self.model.parallel_tiling
@@ -357,44 +421,40 @@ class MiniMaxH3VideoVAE(nn.Module, DistributedVaeMixin):
 
     @torch.inference_mode()
     def decode_latent(self, latent: torch.Tensor) -> torch.Tensor:
-        channels = int(self.config_dict["latent_channels"])
+        with self._decode_tiling_context(latent):
-        mean = torch.tensor(
+            decoded = self.model.decode_base(self._denormalize_latent(latent))
-            self.config_dict["latents_mean"],
+        return self._normalize_decoded_frames(decoded)
-            device=latent.device,
-            dtype=latent.dtype,
-        ).view(1, channels, 1, 1, 1)
-        std = torch.tensor(
-            self.config_dict["latents_std"],
-            device=latent.device,
-            dtype=latent.dtype,
-        ).view(1, channels, 1, 1, 1)
-        # The checkpoint hands rank r the tiles ``range(r, num_tiles, sp_size)``
-        # and then rejects an empty share inside the gather. A rank with no
-        # tiles raises and leaves the collective while the others block in it
-        # forever, so too few tiles hangs the whole stage rather than failing
-        # it. Tile count depends only on the latent shape, so every rank takes
-        # this branch together.
-        num_tiles = self._decoder_tile_count(latent)
-        if self.parallel_size > 1 and num_tiles < self.parallel_size:
-            logger.warning_once(
-                "MiniMax-H3 VAE decode splits into %d tile(s) but the tile group has "
-                "%d ranks; decoding rank-locally for this shape instead, which is "
-                "slower but avoids ranks without tiles hanging the collective.",
-                num_tiles,
-                self.parallel_size,
-            )
-            tiling_context: AbstractContextManager = self._rank_local_tiling()
-        else:
-            tiling_context = nullcontext()
 
-        with tiling_context:
+    @torch.inference_mode()
-            decoded = self.model.decode_base(latent * std + mean)
+    def decode_latent_with_chunks(
-        frames = self.model.processor.revert_tensor(decoded)
+        self,
-        if frames.ndim == 4:
+        latent: torch.Tensor,
-            frames = frames.unsqueeze(0).transpose(1, 2)
+        chunk_callback: MiniMaxH3VideoChunkCallback | None,
-        if frames.ndim != 5:
+    ) -> torch.Tensor:
-            raise ValueError(f"unexpected decoded video shape {tuple(frames.shape)}")
+        """Decode a video while publishing committed temporal chunks.
-        return frames.float()
+
+        Every rank collaborating in the configured H3 VAE tile group must call
+        this method, with ``chunk_callback`` supplied only on group rank 0. A
+        callback failure is deferred until the compatibility decoder finishes its
+        remaining collectives, then propagated to every group rank. With
+        ``parallel_size=1`` the runtime designates each request's output owner
+        by supplying a callback there and ``None`` on replica peers.
+
+        This is a model-local producer seam before the typed media boundary in
+        RFC #6541 / PR #6615. Representation conversion, transport,
+        presentation policy, and encoding remain runtime-owned. Chunks retain
+        the VAE canvas dimensions; a runtime consumer must apply the same
+        requested-size crop as the complete H3 pipeline output.
+        """
+
+        if self._chunk_decode_coordinator is None:
+            self._chunk_decode_coordinator = _MiniMaxH3ChunkDecodeCoordinator()
+        return self._chunk_decode_coordinator.decode(
+            self,
+            latent,
+            chunk_callback,
+            group=self._chunk_process_group(),
+        )
 
 
 class MiniMaxH3AudioVAE(nn.Module):
@@ -516,4 +576,10 @@ class MiniMaxH3AudioVAE(nn.Module):
         return waveform.permute(1, 0, 2).contiguous().float()
 
 
-__all__ = ["MiniMaxH3AudioVAE", "MiniMaxH3VideoVAE"]
+__all__ = [
+    "MiniMaxH3AudioVAE",
+    "MiniMaxH3ChunkCallbackPeerError",
+    "MiniMaxH3ChunkedDecodeUnsupportedError",
+    "MiniMaxH3VideoChunkCallback",
+    "MiniMaxH3VideoVAE",
+]