diff --git a/src/megatron/bridge/data/energon/energon_provider.py b/src/megatron/bridge/data/energon/energon_provider.py
index f33ea48dc1..ded5f8d308 100644
--- a/src/megatron/bridge/data/energon/energon_provider.py
+++ b/src/megatron/bridge/data/energon/energon_provider.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
 from dataclasses import dataclass
 from typing import Any, Optional
 
@@ -33,13 +34,38 @@ class EnergonProvider(DatasetProvider):
     num_workers: int_repr
     dataloader_type: str = "external"
     task_encoder: Optional[Any] = None
-    # Enable batch-level online sequence packing
+    # Existing in-batch packing switch.
+    # Semantics: pack samples *within each already formed micro-batch*.
+    # This path keeps historical behavior used by existing recipes.
     pack_sequences_in_batch: bool = False
+    # THD dataloader switch for batch-level online packing.
+    # Semantics: pack samples *across a dataloader-side candidate buffer*
+    # before a micro-batch is formed.
+    # This mode is independent from pack_sequences_in_batch.
+    batch_level_packing: bool = False
+    packing_buffer_size: Optional[int] = None
+    shuffle_buffer_size: int = 100
+    # Optional bin selector for datasets split into bin directories.
+    # Used to pin data selection and align comparisons with Energon BSHD.
+    cord_bins_root: Optional[str] = None
+    cord_bin_prefix: str = "cord_bin_"
+    cord_bin_id: Optional[str] = None
 
     def build_datasets(self, context: DatasetBuildContext):
-        assert self.path, "EnergonProvider.path must be set. Use CLI override: dataset.path=<path>"
+        resolved_path = self.path
+        if self.cord_bin_id is not None and self.cord_bin_id != "":
+            assert self.cord_bins_root, (
+                "EnergonProvider.cord_bins_root must be set when dataset.cord_bin_id is provided."
+            )
+            resolved_path = os.path.join(self.cord_bins_root, f"{self.cord_bin_prefix}{self.cord_bin_id}")
+
+        assert resolved_path, "EnergonProvider.path must be set. Use CLI override: dataset.path=<path>"
+        if self.task_encoder is not None and hasattr(self.task_encoder, "seq_len"):
+            self.task_encoder.seq_len = self.seq_length
+            self.task_encoder.seq_length = self.seq_length
+        effective_packing_buffer_size = self.packing_buffer_size if self.batch_level_packing else None
         dataset = EnergonMultiModalDataModule(
-            path=self.path,
+            path=resolved_path,
             tokenizer=context.tokenizer if context.tokenizer is not None else self.tokenizer,
             image_processor=self.image_processor,
             seq_length=self.seq_length,
@@ -47,6 +73,8 @@ def build_datasets(self, context: DatasetBuildContext):
             micro_batch_size=self.micro_batch_size,
             global_batch_size=self.global_batch_size,
             num_workers=self.num_workers,
+            packing_buffer_size=effective_packing_buffer_size,
+            shuffle_buffer_size=self.shuffle_buffer_size,
             pg_collection=context.pg_collection,
         )
         return (
diff --git a/src/megatron/bridge/models/conversion/auto_bridge.py b/src/megatron/bridge/models/conversion/auto_bridge.py
index 0c61b75d47..ddb411142c 100644
--- a/src/megatron/bridge/models/conversion/auto_bridge.py
+++ b/src/megatron/bridge/models/conversion/auto_bridge.py
@@ -30,7 +30,6 @@
 
 from megatron.core.transformer.module import MegatronModule
 from megatron.core.transformer.transformer_config import MLATransformerConfig, TransformerConfig
-from modelopt.torch.quantization.utils import is_quantized
 from safetensors.torch import save_file
 from transformers.configuration_utils import PretrainedConfig
 from typing_extensions import Unpack
@@ -51,6 +50,21 @@
 
 logger = logging.getLogger(__name__)
 
+try:
+    from modelopt.torch.quantization.utils import is_quantized
+except ImportError as exc:
+    # modelopt/scipy is optional for recipe import and training startup.
+    # Keep quantization-only export path available when dependency exists.
+    logger.warning(
+        "modelopt quantization utils unavailable; quantized-export detection disabled: %s",
+        exc,
+    )
+
+    def is_quantized(_model: object) -> bool:
+        """Fallback quantization probe when ModelOpt is unavailable."""
+        return False
+
+
 MegatronModelT = TypeVar("MegatronModelT", bound=MegatronModule)
 DataclassT = TypeVar("DataclassT")
 
diff --git a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/model.py b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/model.py
index 81b9e853b1..b753b05e51 100644
--- a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/model.py
+++ b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/model.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 from typing import Optional
 
 import torch
@@ -42,16 +43,23 @@
     collapse_thw,
     get_dist_train_vision_dp_data,
     get_vision_cp_data,
+    is_rank_0,
     pack_dist_train_vision_module_output,
     preprocess_packed_seqs,
     qwen3vl_cp_split,
     reorganize_inputs,
     split_data_cp_rank,
     split_deepstack_embs,
+    thd_diag_align_enabled,
+    thd_diag_enabled,
+    thd_diag_mrope_enabled,
 )
 from megatron.bridge.models.qwen_vl.modelling_qwen3_vl.vision_model import Qwen3VLVisionModel
 
 
+logger = logging.getLogger(__name__)
+
+
 class Qwen3VLModel(MegatronModule):
     """Qwen3VL multi-modal model.
 
@@ -349,6 +357,8 @@ def forward(
         video_input_mask: torch.Tensor = None,
         cp_img_num: list[int] = None,
         images_padded: list[bool] = None,
+        rope_cu_seqlens: torch.Tensor = None,
+        moe_padding_mask: torch.Tensor = None,
         inference_context: object | None = None,
         runtime_gather_output: bool | None = None,
         mm_token_type_ids: torch.Tensor = None,
@@ -399,6 +409,19 @@ def forward(
         # so it must be a real tensor. For packed sequences we use the THD-format
         # input_ids_thd (updated below); for regular sequences we use input_ids as-is.
         lm_input_ids = input_ids
+        moe_padding_mask_for_lm = moe_padding_mask
+
+        def _mask_summary(mask: torch.Tensor, max_items: int = 8) -> tuple[int, str]:
+            flat_idx = torch.nonzero(mask.reshape(-1), as_tuple=False).view(-1)
+            count = int(flat_idx.numel())
+            if count == 0:
+                return count, "[]"
+            head = flat_idx[:max_items].tolist()
+            tail = flat_idx[-max_items:].tolist() if count > max_items else []
+            summary = f"head={head}"
+            if tail:
+                summary += f", tail={tail}"
+            return count, summary
 
         if self.pre_process:
             # can reorganize_inputs at dataset
@@ -506,10 +529,60 @@ def forward(
             if packed_seq_params is not None:
                 if attention_mask is None:
                     attention_mask = torch.ones_like(input_ids, dtype=torch.bool, device=input_ids.device)
+                attn_mask_bool = attention_mask.bool()
                 input_ids_thd, _ = preprocess_packed_seqs(
-                    input_ids, attention_mask, pre_process=True, pg_collection=self.pg_collection
+                    input_ids, attn_mask_bool, pre_process=True, pg_collection=self.pg_collection
                 )
                 lm_input_ids = input_ids_thd
+                if moe_padding_mask_for_lm is not None:
+                    moe_padding_mask_for_lm = preprocess_packed_seqs(
+                        moe_padding_mask_for_lm.to(dtype=torch.int32),
+                        attn_mask_bool,
+                        pre_process=True,
+                        pg_collection=self.pg_collection,
+                    )[0].bool()
+                if thd_diag_align_enabled() and is_rank_0() and labels is not None and loss_mask is not None:
+                    labels_thd = preprocess_packed_seqs(
+                        labels,
+                        attn_mask_bool,
+                        pre_process=True,
+                        pg_collection=self.pg_collection,
+                    )[0]
+                    loss_mask_thd = preprocess_packed_seqs(
+                        loss_mask.to(dtype=torch.float32),
+                        attn_mask_bool,
+                        pre_process=True,
+                        pg_collection=self.pg_collection,
+                    )[0]
+                    labels_valid_pre = labels.ne(-100)
+                    labels_valid_post = labels_thd.ne(-100)
+                    loss_valid_pre = loss_mask > 0
+                    loss_valid_post = loss_mask_thd > 0
+                    label_pre_cnt, label_pre_idx = _mask_summary(labels_valid_pre)
+                    label_post_cnt, label_post_idx = _mask_summary(labels_valid_post)
+                    loss_pre_cnt, loss_pre_idx = _mask_summary(loss_valid_pre)
+                    loss_post_cnt, loss_post_idx = _mask_summary(loss_valid_post)
+                    logger.info(
+                        "[THD_DIAG][align] labels_pre_count=%d labels_post_count=%d labels_same=%s labels_pre_%s labels_post_%s",
+                        label_pre_cnt,
+                        label_post_cnt,
+                        str(bool(torch.equal(labels_valid_pre, labels_valid_post))),
+                        label_pre_idx,
+                        label_post_idx,
+                    )
+                    logger.info(
+                        "[THD_DIAG][align] loss_pre_count=%d loss_post_count=%d loss_same=%s loss_pre_%s loss_post_%s",
+                        loss_pre_cnt,
+                        loss_post_cnt,
+                        str(bool(torch.equal(loss_valid_pre, loss_valid_post))),
+                        loss_pre_idx,
+                        loss_post_idx,
+                    )
+                    logger.info(
+                        "[THD_DIAG][align] pre_label_loss_same=%s post_label_loss_same=%s",
+                        str(bool(torch.equal(labels_valid_pre, loss_valid_pre))),
+                        str(bool(torch.equal(labels_valid_post, loss_valid_post))),
+                    )
                 _, _, vision_mask_thd = reorganize_inputs(
                     input_ids=input_ids_thd,
                     pixel_values=pixel_values,
@@ -530,7 +603,7 @@ def forward(
                         tmp_embeddings[vision_mask] = deepstack_visual_embed
                         tmp_embeddings_thd = preprocess_packed_seqs(
                             tmp_embeddings.contiguous(),
-                            attention_mask,
+                            attn_mask_bool,
                             pre_process=True,
                             pg_collection=self.pg_collection,
                         )[0]
@@ -542,7 +615,7 @@ def forward(
                 combined_embeddings_thd = (
                     preprocess_packed_seqs(
                         combined_embeddings.transpose(0, 1).contiguous(),
-                        attention_mask,
+                        attn_mask_bool,
                         pre_process=True,
                         pg_collection=self.pg_collection,
                     )[0]
@@ -562,9 +635,17 @@ def forward(
             if packed_seq_params is not None:
                 if attention_mask is None:
                     attention_mask = torch.ones_like(input_ids, dtype=torch.bool, device=input_ids.device)
+                attn_mask_bool = attention_mask.bool()
                 lm_input_ids, _ = preprocess_packed_seqs(
-                    input_ids, attention_mask, pre_process=True, pg_collection=self.pg_collection
+                    input_ids, attn_mask_bool, pre_process=True, pg_collection=self.pg_collection
                 )
+                if moe_padding_mask_for_lm is not None:
+                    moe_padding_mask_for_lm = preprocess_packed_seqs(
+                        moe_padding_mask_for_lm.to(dtype=torch.int32),
+                        attn_mask_bool,
+                        pre_process=True,
+                        pg_collection=self.pg_collection,
+                    )[0].bool()
 
         visual_pos_masks = vision_mask
         deepstack_visual_embeds = deepstack_feature_lists
@@ -591,38 +672,186 @@ def forward(
                 )
 
         if position_ids is None:
-            # BSHD
-            # Megatron uses 4D bool masks ([B|1,1,S,S], True=masked); HF uses 2D keep masks ([B,S], 1=keep)
-            # For simplicity, we set hf_attention_mask to None.
-            hf_attention_mask = None
-            position_ids, _ = get_rope_index(
-                self.config.spatial_merge_size,
-                self.image_token_id,
-                self.video_token_id,
-                self.vision_start_token_id,
-                input_ids,
-                image_grid_thw=image_grid_thw,
-                video_grid_thw=video_grid_thw,
-                attention_mask=hf_attention_mask,
-            )  #  [3*b*s]
             if packed_seq_params is not None:
-                # convert position_ids to THD format
+                # Packed sequences: compute MRoPE per sub-sequence independently
+                # so that each sub-sequence's positions restart from 0.
+                cu = rope_cu_seqlens if rope_cu_seqlens is not None else packed_seq_params.cu_seqlens_q
+                if cu.dim() > 1:
+                    cu = cu.squeeze()
+                seq_lens = (cu[1:] - cu[:-1]).tolist()
+                num_seqs = len(seq_lens)
+                max_subseq_len = int(max(seq_lens))
+                total_len = input_ids.shape[1]
+                flat_ids = input_ids.squeeze(0)
+
+                sub_ids = torch.zeros(num_seqs, max_subseq_len, dtype=input_ids.dtype, device=input_ids.device)
+                sub_mask = torch.zeros(num_seqs, max_subseq_len, dtype=torch.int32, device=input_ids.device)
+                for i, sl in enumerate(seq_lens):
+                    start = int(cu[i].item())
+                    sl_int = int(sl)
+                    sub_ids[i, :sl_int] = flat_ids[start : start + sl_int]
+                    sub_mask[i, :sl_int] = 1
+
+                position_ids, _ = get_rope_index(
+                    self.config.spatial_merge_size,
+                    self.image_token_id,
+                    self.video_token_id,
+                    self.vision_start_token_id,
+                    sub_ids,
+                    image_grid_thw=image_grid_thw,
+                    video_grid_thw=video_grid_thw,
+                    attention_mask=sub_mask,
+                )  # [3, num_seqs, max_subseq_len]
+
+                packed_pos = torch.zeros(3, 1, total_len, dtype=position_ids.dtype, device=position_ids.device)
+                for i, sl in enumerate(seq_lens):
+                    start = int(cu[i].item())
+                    sl_int = int(sl)
+                    packed_pos[:, 0, start : start + sl_int] = position_ids[:, i, :sl_int]
+
+                if thd_diag_mrope_enabled() and is_rank_0():
+                    # Summarize per-subsequence MRoPE position behavior before THD remap.
+                    # This is diagnostics-only and intentionally does not affect semantics.
+                    show_n = 4
+                    head_stats = []
+                    for i, sl in enumerate(seq_lens[:show_n]):
+                        start = int(cu[i].item())
+                        sl_int = int(sl)
+                        if sl_int <= 0:
+                            head_stats.append(f"{i}:len=0")
+                            continue
+                        pos0 = packed_pos[0, 0, start : start + sl_int]
+                        nonmono = int((pos0[1:] < pos0[:-1]).sum().item()) if sl_int > 1 else 0
+                        head_stats.append(
+                            f"{i}:len={sl_int},start={int(pos0[0].item())},end={int(pos0[-1].item())},"
+                            f"min={int(pos0.min().item())},max={int(pos0.max().item())},dec={nonmono}"
+                        )
+                    logger.info(
+                        "[THD_DIAG][mrope] pre_thd cp_size=%d num_seqs=%d total_len=%d seq_lens_head=%s seg_stats_head=%s",
+                        int(cp_size),
+                        int(num_seqs),
+                        int(total_len),
+                        seq_lens[:show_n],
+                        "; ".join(head_stats),
+                    )
+
+                position_ids = packed_pos
+
+                attn_mask_bool = attention_mask.bool()
                 position_ids = (
                     preprocess_packed_seqs(
                         position_ids.permute(1, 2, 0),
-                        attention_mask,
+                        attn_mask_bool,
                         pre_process=True,
                         pg_collection=self.pg_collection,
                     )[0]
                     .permute(2, 0, 1)
                     .contiguous()
                 )
+                if thd_diag_mrope_enabled() and is_rank_0():
+                    pos0_post = position_ids[0, 0]
+                    post_len = int(pos0_post.numel())
+                    head_vals = pos0_post[:12].tolist()
+                    tail_vals = pos0_post[-12:].tolist() if post_len > 12 else []
+                    nonmono_post = int((pos0_post[1:] < pos0_post[:-1]).sum().item()) if post_len > 1 else 0
+                    logger.info(
+                        "[THD_DIAG][mrope] post_thd cp_size=%d len=%d min=%d max=%d dec=%d head=%s tail=%s",
+                        int(cp_size),
+                        post_len,
+                        int(pos0_post.min().item()) if post_len > 0 else -1,
+                        int(pos0_post.max().item()) if post_len > 0 else -1,
+                        nonmono_post,
+                        head_vals,
+                        tail_vals,
+                    )
+                    # Strict check: verify THD-remapped position_ids equals explicit bool-mask gather
+                    # semantics for cp_size==1. This is diagnostics-only.
+                    if int(cp_size) == 1:
+                        tp_size = int(self.pg_collection.tp.size())
+                        align_size = tp_size
+                        packed_pos_bsd = packed_pos.permute(1, 2, 0).contiguous()  # [1, S, 3]
+                        valid_len = int(attn_mask_bool[0].sum().item())
+                        padded_len = valid_len + ((align_size - (valid_len % align_size)) % align_size)
+                        expected_bsd = torch.zeros(
+                            (1, padded_len, packed_pos_bsd.size(-1)),
+                            dtype=packed_pos_bsd.dtype,
+                            device=packed_pos_bsd.device,
+                        )
+                        expected_bsd[0, :valid_len] = packed_pos_bsd[0, attn_mask_bool[0]]
+                        expected_pid = expected_bsd.permute(2, 0, 1).contiguous()
+
+                        same = bool(torch.equal(position_ids, expected_pid))
+                        mismatch = (position_ids != expected_pid).any(dim=0).squeeze(0)
+                        mismatch_cnt = int(mismatch.sum().item())
+                        mismatch_idx = torch.nonzero(mismatch, as_tuple=False).flatten()
+                        head_idx = mismatch_idx[:8].tolist()
+                        tail_idx = mismatch_idx[-8:].tolist() if mismatch_idx.numel() > 8 else []
+                        logger.info(
+                            "[THD_DIAG][mrope] strict_match=%s mismatch_count=%d expected_len=%d actual_len=%d mismatch_head=%s mismatch_tail=%s",
+                            str(same),
+                            mismatch_cnt,
+                            int(expected_pid.size(-1)),
+                            int(position_ids.size(-1)),
+                            head_idx,
+                            tail_idx,
+                        )
+                    else:
+                        logger.info(
+                            "[THD_DIAG][mrope] strict_match_skipped cp_size=%d (currently only checks cp_size==1)",
+                            int(cp_size),
+                        )
                 attention_mask = None
                 self.language_model.rotary_pos_emb.is_thd_format = True
+            else:
+                # BSHD
+                # Megatron uses 4D bool masks ([B|1,1,S,S], True=masked); HF uses 2D keep masks ([B,S], 1=keep)
+                # For get_rope_index we pass None to avoid semantic mismatch.
+                hf_attention_mask = None
+                position_ids, _ = get_rope_index(
+                    self.config.spatial_merge_size,
+                    self.image_token_id,
+                    self.video_token_id,
+                    self.vision_start_token_id,
+                    input_ids,
+                    image_grid_thw=image_grid_thw,
+                    video_grid_thw=video_grid_thw,
+                    attention_mask=hf_attention_mask,
+                )  #  [3*b*s]
 
         torch.cuda.nvtx.range_pop()
         torch.cuda.nvtx.range_push("Qwen3VLModel.forward.language_model")
 
+        # For THD packed path we intentionally keep attention_mask=None for model forward.
+        # MoE aux/global-aux accounting still needs to ignore tail padding tokens, so pass
+        # a dedicated padding_mask (True=padding) to GPTModel/Router.
+        padding_mask_for_moe = None
+        if packed_seq_params is not None and lm_input_ids is not None:
+            if moe_padding_mask_for_lm is not None:
+                padding_mask_for_moe = moe_padding_mask_for_lm.bool()
+                mask_source = "explicit"
+            else:
+                # Fallback for old call sites that do not provide explicit packed padding mask.
+                padding_mask_for_moe = lm_input_ids.eq(0)
+                mask_source = "token_eq_0_fallback"
+            if thd_diag_enabled() and is_rank_0():
+                input_zero_cnt = int(input_ids.eq(0).sum().item()) if input_ids is not None else -1
+                lm_zero_cnt = int(lm_input_ids.eq(0).sum().item())
+                pad_cnt = int(padding_mask_for_moe.sum().item())
+                tok_cnt = int(padding_mask_for_moe.numel())
+                loss_valid = int((loss_mask > 0).sum().item()) if loss_mask is not None else -1
+                loss_total = int(loss_mask.numel()) if loss_mask is not None else -1
+                logger.info(
+                    "[THD_DIAG][model] padding_mask_for_moe: source=%s total_tokens=%d padding_tokens=%d valid_tokens=%d input_zero_tokens=%d lm_zero_tokens=%d loss_valid_tokens=%d loss_total_tokens=%d",
+                    mask_source,
+                    tok_cnt,
+                    pad_cnt,
+                    tok_cnt - pad_cnt,
+                    input_zero_cnt,
+                    lm_zero_cnt,
+                    loss_valid,
+                    loss_total,
+                )
+
         output = self.language_model(
             input_ids=lm_input_ids,
             position_ids=position_ids,  # None in encoder
@@ -630,8 +859,9 @@ def forward(
             decoder_input=combined_embeddings,  # only not None in the first decoder PP stage
             labels=labels,  # only not None in the last decoder PP stage
             loss_mask=loss_mask,  # Added for THD training compatibility
-            inference_params=inference_params,  # currently always None
-            packed_seq_params=packed_seq_params,  # currently always None
+            padding_mask=padding_mask_for_moe,  # for MoE routing/aux-loss token accounting
+            inference_params=inference_params,  # training path keeps this as None
+            packed_seq_params=packed_seq_params,
             visual_pos_masks=visual_pos_masks,
             deepstack_visual_embeds=deepstack_visual_embeds,
             **(extra_block_kwargs or {}),
diff --git a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/text_model.py b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/text_model.py
index d067655a1b..d6781ced3a 100644
--- a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/text_model.py
+++ b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/text_model.py
@@ -120,6 +120,7 @@ def forward(
         *,
         inference_params: Optional[BaseInferenceContext] = None,
         loss_mask: Optional[Tensor] = None,
+        padding_mask: Optional[Tensor] = None,
         # args for deepstack
         visual_pos_masks: Optional[torch.Tensor] = None,
         deepstack_visual_embeds: Optional[list[torch.Tensor]] = None,
@@ -148,6 +149,7 @@ def forward(
             decoder_input=decoder_input,
             inference_context=inference_context,
             packed_seq_params=packed_seq_params,
+            padding_mask=padding_mask,
         )
 
         (
@@ -156,7 +158,8 @@ def forward(
             rotary_pos_cos,
             rotary_pos_sin,
             sequence_len_offset,
-        ) = preproc_output[:5]
+            padding_mask,
+        ) = preproc_output[:6]
 
         # Run decoder.
         hidden_states = self.decoder(
@@ -170,6 +173,7 @@ def forward(
             # the standard components only.
             packed_seq_params=packed_seq_params,
             sequence_len_offset=sequence_len_offset,
+            padding_mask=padding_mask,
             visual_pos_masks=visual_pos_masks,
             deepstack_visual_embeds=deepstack_visual_embeds,
             **(extra_block_kwargs or {}),
@@ -204,6 +208,7 @@ def _sp_scatter_embedding(input_ids, position_ids):
             loss_mask=loss_mask,
             decoder_input=decoder_input,
             attention_mask=attention_mask,
+            padding_mask=padding_mask,
             inference_params=inference_params,
             packed_seq_params=packed_seq_params,
             sequence_len_offset=sequence_len_offset,
diff --git a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/transformer_block.py b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/transformer_block.py
index 6284617bed..440e45c873 100644
--- a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/transformer_block.py
+++ b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/transformer_block.py
@@ -503,6 +503,7 @@ def _checkpointed_forward(
         rotary_pos_emb: Tensor,
         attention_bias: Tensor,
         packed_seq_params: PackedSeqParams,
+        padding_mask: Optional[Tensor],
         use_inner_fp8_context: bool,
         # args for deepstack
         visual_pos_masks: Optional[torch.Tensor] = None,
@@ -517,6 +518,7 @@ def custom_forward(
                 context,
                 context_mask,
                 rotary_pos_emb,
+                padding_mask,
                 visual_pos_masks,
                 *deepstack_visual_embeds_args,
             ):
@@ -538,6 +540,7 @@ def custom_forward(
                             attention_bias=attention_bias,
                             inference_context=None,
                             packed_seq_params=packed_seq_params,
+                            padding_mask=padding_mask,
                         )
 
                         if self.pre_process and deepstack_visual_embeds is not None:
@@ -567,6 +570,7 @@ def checkpoint_handler(forward_func):
                     context,
                     context_mask,
                     rotary_pos_emb,
+                    padding_mask,
                     visual_pos_masks,
                     *deepstack_visual_embeds_tuple,
                 )
@@ -579,6 +583,7 @@ def checkpoint_handler(forward_func):
                     context,
                     context_mask,
                     rotary_pos_emb,
+                    padding_mask,
                     visual_pos_masks,
                     *deepstack_visual_embeds_tuple,
                 )
@@ -618,6 +623,7 @@ def checkpoint_handler(forward_func):
                         context,
                         context_mask,
                         rotary_pos_emb,
+                        padding_mask,
                         visual_pos_masks,
                         *deepstack_visual_embeds_tuple,
                     )
@@ -639,6 +645,7 @@ def forward(
         inference_context: Optional[BaseInferenceContext] = None,
         packed_seq_params: Optional[PackedSeqParams] = None,
         sequence_len_offset: Optional[Tensor] = None,
+        padding_mask: Optional[Tensor] = None,
         *,
         inference_params: Optional[BaseInferenceContext] = None,
         # args for deepstack
@@ -730,6 +737,7 @@ def forward(
                     rotary_pos_emb=rotary_pos_emb,
                     attention_bias=attention_bias,
                     packed_seq_params=packed_seq_params,
+                    padding_mask=padding_mask,
                     use_inner_fp8_context=use_inner_fp8_context,
                     visual_pos_masks=visual_pos_masks,
                     deepstack_visual_embeds=deepstack_visual_embeds,
@@ -754,6 +762,7 @@ def forward(
                             inference_context=inference_context,
                             packed_seq_params=packed_seq_params,
                             sequence_len_offset=sequence_len_offset,
+                            padding_mask=padding_mask,
                         )
 
                         if self.pre_process and deepstack_visual_embeds is not None:
diff --git a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/utils.py b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/utils.py
index 35b40d3ee3..9f87bb96a0 100644
--- a/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/utils.py
+++ b/src/megatron/bridge/models/qwen_vl/modelling_qwen3_vl/utils.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 
+import os
 from dataclasses import dataclass
 from typing import Optional, Sequence, Union
 
@@ -26,6 +27,22 @@
 from torch import nn
 
 from megatron.bridge.models.qwen_vl.modelling_qwen3_vl.transformer_config import Qwen3VLTransformerConfig
+from megatron.bridge.utils.common_utils import is_rank_0  # noqa: F401
+
+
+def thd_diag_enabled() -> bool:
+    """Return whether THD diagnostics are enabled."""
+    return os.environ.get("THD_DIAG", "0") not in ("0", "", "false", "False")
+
+
+def thd_diag_align_enabled() -> bool:
+    """Return whether THD alignment diagnostics are enabled."""
+    return os.environ.get("THD_DIAG_ALIGN", "0") not in ("0", "", "false", "False")
+
+
+def thd_diag_mrope_enabled() -> bool:
+    """Return whether THD MRoPE diagnostics are enabled."""
+    return os.environ.get("THD_DIAG_MROPE", "0") not in ("0", "", "false", "False")
 
 
 # copied from https://github.com/huggingface/transformers/blob/main/src/transformers/models/qwen3_vl/modeling_qwen3_vl.py
@@ -694,6 +711,11 @@ def preprocess_packed_seqs(
     See https://github.com/NVIDIA/TransformerEngine/issues/1368
     """
     batch_size = input_ids.shape[0]
+    # Boolean mask is required for advanced indexing semantics below:
+    # input_ids[i, attention_mask[i]]
+    # If mask is int (0/1), PyTorch treats it as positional index selection
+    # rather than keep-mask, which can silently corrupt THD remapping.
+    attention_mask = attention_mask.bool()
 
     # Ensure boolean dtype for correct advanced indexing (bool → mask select,
     # int → fancy index which silently corrupts data when values are 0/1).
diff --git a/src/megatron/bridge/models/qwen_vl/qwen3_vl_step.py b/src/megatron/bridge/models/qwen_vl/qwen3_vl_step.py
index c87202262b..1a044a06dd 100644
--- a/src/megatron/bridge/models/qwen_vl/qwen3_vl_step.py
+++ b/src/megatron/bridge/models/qwen_vl/qwen3_vl_step.py
@@ -36,6 +36,7 @@
 
 
 logger = logging.getLogger(__name__)
+_BATCH_LEVEL_PACKING_WARNED = False
 
 
 def get_batch_from_iterator(
@@ -248,7 +249,19 @@ def forward_step(
 
     # To be compatible with qwen3vl, we move the sequence padding and packing to forward_step function.
     # Qwen3VL model need the original input and do cp and sp split in model.forward.
-    pack_sequences_in_batch = getattr(state.cfg.dataset, "pack_sequences_in_batch", False)
+    in_batch_pack_enabled = getattr(state.cfg.dataset, "pack_sequences_in_batch", False)
+    batch_level_pack_enabled = getattr(state.cfg.dataset, "batch_level_packing", False)
+    # qwen3_vl_step does not yet wire full THD metadata (rope/moe padding-mask)
+    # for batch-level packing. Keep this path disabled to avoid incorrect token
+    # accounting until parity with training.vlm_step is implemented.
+    global _BATCH_LEVEL_PACKING_WARNED
+    if batch_level_pack_enabled and not _BATCH_LEVEL_PACKING_WARNED:
+        logger.warning(
+            "dataset.batch_level_packing is not fully supported in qwen3_vl_step yet; "
+            "falling back to in-batch packing behavior."
+        )
+        _BATCH_LEVEL_PACKING_WARNED = True
+    enable_sequence_packing = in_batch_pack_enabled
 
     tokens, labels, loss_mask, attention_mask, position_ids, packed_seq_params = pack_or_pad_batch_sequences(
         tokens,
@@ -275,7 +288,7 @@ def forward_step(
     forward_args["input_ids"] = original_tokens
     # calculate position_ids in model forward
     forward_args["position_ids"] = None
-    if pack_sequences_in_batch:
+    if enable_sequence_packing:
         if forward_args["labels"] is not None:
             # When using pp, labels could be None
             forward_args["labels"] = forward_args["labels"].reshape(1, -1)
diff --git a/src/megatron/bridge/recipes/__init__.py b/src/megatron/bridge/recipes/__init__.py
index 46d51fdb64..3685045ec7 100644
--- a/src/megatron/bridge/recipes/__init__.py
+++ b/src/megatron/bridge/recipes/__init__.py
@@ -18,8 +18,18 @@
 This module exposes all recipe configurations from all model families.
 """
 
-from megatron.bridge.diffusion.recipes.flux.flux import *
-from megatron.bridge.diffusion.recipes.wan.wan import *
+# Diffusion recipes require optional dependencies (e.g. diffusers). Keep
+# non-diffusion training paths (LLM/VLM) usable when those deps are absent.
+try:
+    from megatron.bridge.diffusion.recipes.flux.flux import *
+except ModuleNotFoundError:
+    pass
+
+try:
+    from megatron.bridge.diffusion.recipes.wan.wan import *
+except ModuleNotFoundError:
+    pass
+
 from megatron.bridge.recipes.deepseek import *
 from megatron.bridge.recipes.gemma import *
 from megatron.bridge.recipes.gemma3_vl import *
diff --git a/src/megatron/bridge/recipes/qwen_vl/__init__.py b/src/megatron/bridge/recipes/qwen_vl/__init__.py
index 9bc536c45d..93dc981f59 100644
--- a/src/megatron/bridge/recipes/qwen_vl/__init__.py
+++ b/src/megatron/bridge/recipes/qwen_vl/__init__.py
@@ -52,6 +52,8 @@
     qwen35_vl_35b_a3b_peft_config,
     qwen35_vl_35b_a3b_pretrain_mock_config,
     qwen35_vl_35b_a3b_sft_config,
+    qwen35_vl_35b_a3b_sft_energon_config,
+    qwen35_vl_35b_a3b_sft_energon_nopack_config,
     qwen35_vl_122b_a10b_peft_config,
     qwen35_vl_122b_a10b_pretrain_mock_config,
     qwen35_vl_122b_a10b_sft_config,
@@ -77,6 +79,8 @@
     "qwen35_vl_27b_sft_config",
     # Qwen3.5-VL SFT configs — MoE
     "qwen35_vl_35b_a3b_sft_config",
+    "qwen35_vl_35b_a3b_sft_energon_config",
+    "qwen35_vl_35b_a3b_sft_energon_nopack_config",
     "qwen35_vl_35b_a3b_fsdp_sft_config",
     "qwen35_vl_122b_a10b_sft_config",
     "qwen35_vl_397b_a17b_sft_config",
diff --git a/src/megatron/bridge/recipes/qwen_vl/data/energon/task_encoder.py b/src/megatron/bridge/recipes/qwen_vl/data/energon/task_encoder.py
index 50178a0749..12ec6d9f15 100644
--- a/src/megatron/bridge/recipes/qwen_vl/data/energon/task_encoder.py
+++ b/src/megatron/bridge/recipes/qwen_vl/data/energon/task_encoder.py
@@ -12,16 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import bisect
 import dataclasses
 import logging
+import os
 import re
 from collections import defaultdict
-from dataclasses import dataclass
-from typing import Dict, List
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Union
 
 import numpy as np
 import torch
-from megatron.energon import Batch, DefaultTaskEncoder
+from megatron.energon import Batch, DefaultTaskEncoder, stateless
 from transformers import BatchEncoding
 
 from megatron.bridge.data.energon.task_encoder_utils import (
@@ -39,6 +41,69 @@
 from megatron.bridge.training.utils.visual_inputs import Qwen2_5_VLVisualInputs
 
 
+logger = logging.getLogger(__name__)
+_LOW_CONTENT_LOG_COUNT = 0
+
+
+def _thd_diag_enabled() -> bool:
+    return os.environ.get("THD_DIAG", "0") not in ("0", "", "false", "False")
+
+
+def _thd_diag_low_content_threshold() -> int:
+    raw = os.environ.get("THD_DIAG_LOW_CONTENT_TOKENS", "2048")
+    try:
+        return max(0, int(raw))
+    except ValueError:
+        return 2048
+
+
+def _thd_diag_max_low_content_logs() -> int:
+    raw = os.environ.get("THD_DIAG_MAX_LOW_CONTENT_LOGS", "100")
+    try:
+        return max(0, int(raw))
+    except ValueError:
+        return 100
+
+
+def _search_for_fit(numbers: List[int], capacity: int) -> int:
+    """Binary search for the largest number that fits within capacity."""
+    index = bisect.bisect(numbers, capacity)
+    return -1 if index == 0 else (index - 1)
+
+
+def greedy_knapsack(item_sizes: List[int], samples: List, max_capacity: int) -> List:
+    """Greedy bin-packing with binary search.
+
+    Sorts samples by length ascending, then greedily fills each bin by picking
+    the largest item that still fits (via binary search). Returns a list of bins,
+    each bin being a list of samples.
+    """
+    assert len(item_sizes) == len(samples)
+    if not item_sizes:
+        return []
+
+    sorted_sizes, sorted_samples = zip(*sorted(zip(item_sizes, samples), key=lambda x: x[0]))
+    sorted_sizes = list(sorted_sizes)
+    sorted_samples = list(sorted_samples)
+
+    if sorted_sizes[-1] > max_capacity:
+        raise ValueError(f"Sample size {sorted_sizes[-1]} exceeds max_capacity {max_capacity}")
+
+    knapsacks = []
+    while sorted_sizes:
+        current_knapsack = []
+        remaining = max_capacity
+        while True:
+            idx = _search_for_fit(sorted_sizes, remaining)
+            if idx == -1:
+                break
+            remaining -= sorted_sizes[idx]
+            sorted_sizes.pop(idx)
+            current_knapsack.append(sorted_samples.pop(idx))
+        knapsacks.append(current_knapsack)
+    return knapsacks
+
+
 def process_vision(
     processor, images, videos, fps=None, model_version: str = "qwen-vl", min_pixels=None, max_pixels=None
 ):
@@ -71,19 +136,25 @@ def process_vision(
 
 
 def _resolve_hf_mm_token_ids(hf_tokenizer):
-    """Resolve HF tokenizer ids for <image> and <video> tokens without nemo constants."""
+    """Resolve HF tokenizer ids for ``<image>`` and ``<video>`` tokens strictly."""
 
-    def _get(token_str: str, default_id: int) -> int:
+    def _get(token_str: str) -> int:
         token_attr = getattr(hf_tokenizer, f"{token_str.strip('<>')}_token_id", None)
         if token_attr is not None:
             return int(token_attr)
         try:
-            return int(hf_tokenizer.convert_tokens_to_ids(token_str))
-        except Exception:
-            return default_id
-
-    image_id = _get("<image>", 151655)
-    video_id = _get("<video>", 151656)
+            token_id = hf_tokenizer.convert_tokens_to_ids(token_str)
+        except Exception as exc:
+            raise ValueError(f"Failed to resolve token id for {token_str!r}") from exc
+        if token_id is None:
+            raise ValueError(f"Tokenizer returned None for {token_str!r}")
+        token_id = int(token_id)
+        if token_id < 0:
+            raise ValueError(f"Tokenizer returned invalid id {token_id} for {token_str!r}")
+        return token_id
+
+    image_id = _get("<image>")
+    video_id = _get("<video>")
     return image_id, video_id
 
 
@@ -103,12 +174,45 @@ class QwenVLTaskSample:
     video_input_mask: torch.Tensor
     text: torch.Tensor
     target: torch.Tensor
+    total_len: int = 0
+
+
+@dataclass
+class QwenVLTaskSamplePacked:
+    """Packed sample: multiple QwenVLTaskSample concatenated into one sequence."""
+
+    __key__: str
+    __subflavors__: Dict
+    tokens: torch.Tensor
+    target: torch.Tensor
+    imgs: list
+    videos: list
+    image_thw_grids: list
+    video_thw_grids: list
+    image_input_mask: torch.Tensor
+    video_input_mask: torch.Tensor
+    cu_lengths: torch.Tensor
+    max_length: int
+    num_sub_samples: int
+    sub_sample_lengths: list = field(default_factory=list)
+    num_image_tokens: int = 0
+    num_video_tokens: int = 0
+    num_text_tokens: int = 0
+    num_vit_patches: int = 0
 
 
 @dataclass
 class QwenVLTaskBatch(Batch):
-    """Encoded Batch Format For QwenVL"""
+    """Encoded Batch Format For QwenVL.
+
+    Energon 7.x ``Batch`` requires ``__key__`` and ``__restore_key__`` as
+    keyword-only init fields.  We keep ``__keys__`` (plural) for downstream
+    consumption but satisfy the parent via ``__key__`` (joined) and
+    ``__restore_key__`` (empty tuple placeholder).
+    """
 
+    __key__: str
+    __restore_key__: tuple
     __keys__: List[str]
     __subflavors__: List[Dict]
     # (num_tiles, c, h, w)
@@ -125,6 +229,10 @@ class QwenVLTaskBatch(Batch):
     position_ids: torch.Tensor
     labels: torch.Tensor
     loss_mask: torch.Tensor
+    # Packing metadata (set when using Energon packing)
+    cu_lengths: Optional[torch.Tensor] = None
+    max_lengths: Optional[torch.Tensor] = None
+    cu_seqlens_argmin: Optional[torch.Tensor] = None
 
 
 def convert_to_qwenvl_content(user_input: str, image_pattern: str = "<image>", video_pattern: str = "<video>"):
@@ -162,12 +270,14 @@ def __init__(
         self,
         tokenizer,
         image_processor,
-        temporal_patch_size: int = 2,
-        spatial_merge_size: int = 2,
-        patch_size: int = 14,
-        max_padding_length: int = 4096,
-        min_pixels: int = 200704,
-        max_pixels: int = 1003520,
+        temporal_patch_size: int,
+        spatial_merge_size: int,
+        patch_size: int,
+        image_token_id: int,
+        video_token_id: int,
+        max_padding_length: int,
+        min_pixels: int,
+        max_pixels: int,
     ):
         super().__init__()
 
@@ -182,7 +292,12 @@ def __init__(
         self.patch_size = patch_size
 
         self.seq_len = max_padding_length
-        self.image_token_id, self.video_token_id = _resolve_hf_mm_token_ids(self.hf_tokenizer)
+        self.image_token_id = int(image_token_id)
+        self.video_token_id = int(video_token_id)
+        if self.image_token_id < 0 or self.video_token_id < 0:
+            raise ValueError(
+                f"image_token_id/video_token_id must be non-negative, got {self.image_token_id}/{self.video_token_id}"
+            )
 
     def encode_sample(self, sample: ChatMLSample):
         """
@@ -339,55 +454,342 @@ def encode_sample(self, sample: ChatMLSample):
             video_input_mask=video_input_mask,
             text=torch.from_numpy(final_input_ids),
             target=torch.from_numpy(target),
+            total_len=int(target_length),
         )
 
-    def batch(self, samples: List[QwenVLTaskSample]) -> QwenVLTaskBatch:
+    def select_samples_to_pack(self, samples: List[QwenVLTaskSample]) -> List[List[QwenVLTaskSample]]:
+        """Select groups of samples to pack together using greedy knapsack.
+
+        Called by Energon's PackingDataset. Returns list of bins, each bin
+        is a list of samples whose total length <= self.seq_len.
         """
-        Put encoded sample into Batch, do padding, add labels and visual input masks
+        lengths = [s.total_len for s in samples]
+        return greedy_knapsack(lengths, samples, self.seq_len)
 
-        Args:
-            samples: List of encoded samples
+    @stateless
+    def pack_selected_samples(self, samples: List[QwenVLTaskSample]) -> QwenVLTaskSamplePacked:
+        """Concatenate multiple QwenVLTaskSample into one packed sample.
 
-        Returns:
-            Batch with necessary fields
+        Called by Energon's PackingDataset after select_samples_to_pack.
+        Produces a QwenVLTaskSamplePacked with cu_lengths for THD attention.
         """
-        imgs, image_thw_grids = [], []
+        all_tokens, all_targets = [], []
+        all_imgs, all_videos = [], []
+        all_image_thw_grids, all_video_thw_grids = [], []
+        all_image_masks, all_video_masks = [], []
+        sub_lengths = []
+        total_image_tokens = 0
+        total_video_tokens = 0
+        total_text_tokens = 0
+        total_vit_patches = 0
+
         for s in samples:
+            seq_len = len(s.text)
+            sub_lengths.append(seq_len)
+            all_tokens.append(s.text)
+            all_targets.append(s.target)
+            all_image_masks.append(s.image_input_mask)
+            all_video_masks.append(s.video_input_mask)
+
+            n_img = int(s.image_input_mask.sum().item())
+            n_vid = int(s.video_input_mask.sum().item())
+            n_txt = seq_len - n_img - n_vid
+            total_image_tokens += n_img
+            total_video_tokens += n_vid
+            total_text_tokens += n_txt
+
             if len(s.imgs) > 0:
-                s_imgs = [img for img in s.imgs.unsqueeze(0)]
-                cat_imgs = torch.cat([img for img in s_imgs])
-                imgs.append(cat_imgs)
+                if isinstance(s.imgs, torch.Tensor):
+                    all_imgs.append(s.imgs.unsqueeze(0) if s.imgs.dim() == 3 else s.imgs)
+                else:
+                    all_imgs.append(s.imgs)
             if len(s.image_thw_grids) > 0:
-                s_image_thw_grids = [thw_grids for thw_grids in s.image_thw_grids]
-                image_thw_grids.extend(s_image_thw_grids)
-        videos, video_thw_grids = [], []
-        for s in samples:
+                thw = s.image_thw_grids
+                if isinstance(thw, np.ndarray):
+                    for row in thw:
+                        all_image_thw_grids.append(row)
+                        total_vit_patches += int(np.prod(row))
+                elif isinstance(thw, torch.Tensor):
+                    for row in thw:
+                        all_image_thw_grids.append(row.numpy())
+                        total_vit_patches += int(row.prod().item())
+                else:
+                    all_image_thw_grids.extend(thw)
+
             if len(s.videos) > 0:
-                s_videos = [video for video in s.videos.unsqueeze(0)]
-                cat_videos = torch.cat([video for video in s_videos])
-                videos.append(cat_videos)
+                if isinstance(s.videos, torch.Tensor):
+                    all_videos.append(s.videos.unsqueeze(0) if s.videos.dim() == 3 else s.videos)
+                else:
+                    all_videos.append(s.videos)
             if len(s.video_thw_grids) > 0:
-                s_video_thw_grids = [thw_grids for thw_grids in s.video_thw_grids]
-                video_thw_grids.extend(s_video_thw_grids)
-                # assert s_video_thw_grids.prod(dim=-1).sum() == s_videos.shape[0]
+                thw = s.video_thw_grids
+                if isinstance(thw, np.ndarray):
+                    for row in thw:
+                        all_video_thw_grids.append(row)
+                        total_vit_patches += int(np.prod(row))
+                elif isinstance(thw, torch.Tensor):
+                    for row in thw:
+                        all_video_thw_grids.append(row.numpy())
+                        total_vit_patches += int(row.prod().item())
+                else:
+                    all_video_thw_grids.extend(thw)
+
+        packed_tokens = torch.cat(all_tokens, dim=0)
+        packed_targets = torch.cat(all_targets, dim=0)
+        packed_image_mask = torch.cat(all_image_masks, dim=0)
+        packed_video_mask = torch.cat(all_video_masks, dim=0)
+
+        cu_lengths_list = [0]
+        for length in sub_lengths:
+            cu_lengths_list.append(cu_lengths_list[-1] + length)
+        cu_lengths = torch.tensor(cu_lengths_list, dtype=torch.int32)
+
+        total_packed = int(cu_lengths[-1].item())
+        max_sub_len = max(sub_lengths)
+
+        if _thd_diag_enabled():
+            logger.info(
+                f"[PackingStats] packed {len(samples)} samples → {total_packed} tokens "
+                f"(seq_len={self.seq_len}, utilization={total_packed / self.seq_len * 100:.1f}%), "
+                f"sub_lengths={sub_lengths}, "
+                f"image_tokens={total_image_tokens}, video_tokens={total_video_tokens}, "
+                f"text_tokens={total_text_tokens}, vit_patches={total_vit_patches}"
+            )
+
+        return QwenVLTaskSamplePacked(
+            __key__="+".join(s.__key__ for s in samples),
+            __subflavors__=samples[0].__subflavors__,
+            tokens=packed_tokens,
+            target=packed_targets,
+            imgs=all_imgs,
+            videos=all_videos,
+            image_thw_grids=all_image_thw_grids,
+            video_thw_grids=all_video_thw_grids,
+            image_input_mask=packed_image_mask,
+            video_input_mask=packed_video_mask,
+            cu_lengths=cu_lengths,
+            max_length=max_sub_len,
+            num_sub_samples=len(samples),
+            sub_sample_lengths=sub_lengths,
+            num_image_tokens=total_image_tokens,
+            num_video_tokens=total_video_tokens,
+            num_text_tokens=total_text_tokens,
+            num_vit_patches=total_vit_patches,
+        )
+
+    @staticmethod
+    def _normalize_visual_block(block: torch.Tensor) -> torch.Tensor:
+        """Normalize visual block shape before batch concat.
+
+        Expected downstream format is either:
+        - [N, C, H, W] (patch tiles), or
+        - [N, D] (flattened patch features).
+        """
+        if not isinstance(block, torch.Tensor):
+            raise TypeError(f"Visual block must be torch.Tensor, got {type(block)}")
+
+        # Common case for processor output: [1, N, D] -> [N, D]
+        if block.dim() == 3 and block.shape[0] == 1:
+            return block.squeeze(0)
+
+        # Single tile CHW -> add batch dimension.
+        if block.dim() == 3 and block.shape[0] in (1, 3):
+            return block.unsqueeze(0)
+
+        # Keep already batched formats as-is.
+        if block.dim() in (2, 4):
+            return block
+
+        return block
+
+    @staticmethod
+    def _concat_visual_blocks(blocks: List[torch.Tensor], name: str) -> Optional[torch.Tensor]:
+        """Concatenate visual blocks along first dimension with shape diagnostics."""
+        if not blocks:
+            return None
 
-        # use the max sample lengths in the batch.
+        normalized = [QwenVLTaskEncoder._normalize_visual_block(b) for b in blocks]
+        try:
+            return torch.cat(normalized, dim=0)
+        except RuntimeError as exc:
+            shapes = [tuple(t.shape) for t in normalized]
+            logger.error(f"Failed to concat {name} blocks, shapes={shapes}, error={exc}")
+            raise
+
+    def _collect_visual_data(self, samples):
+        """Gather pixel_values and grid_thw from a list of samples (packed or unpacked)."""
+        imgs, image_thw_grids = [], []
+        videos, video_thw_grids = [], []
+
+        for s in samples:
+            if isinstance(s, QwenVLTaskSamplePacked):
+                for img_block in s.imgs:
+                    if isinstance(img_block, torch.Tensor):
+                        imgs.append(self._normalize_visual_block(img_block))
+                for thw in s.image_thw_grids:
+                    image_thw_grids.append(np.asarray(thw))
+                for vid_block in s.videos:
+                    if isinstance(vid_block, torch.Tensor):
+                        videos.append(self._normalize_visual_block(vid_block))
+                for thw in s.video_thw_grids:
+                    video_thw_grids.append(np.asarray(thw))
+            else:
+                if len(s.imgs) > 0:
+                    s_imgs = self._normalize_visual_block(s.imgs) if isinstance(s.imgs, torch.Tensor) else s.imgs
+                    if isinstance(s_imgs, torch.Tensor):
+                        imgs.append(s_imgs)
+                if len(s.image_thw_grids) > 0:
+                    if isinstance(s.image_thw_grids, np.ndarray):
+                        for row in s.image_thw_grids:
+                            image_thw_grids.append(row)
+                    else:
+                        image_thw_grids.extend(s.image_thw_grids)
+                if len(s.videos) > 0:
+                    s_vids = self._normalize_visual_block(s.videos) if isinstance(s.videos, torch.Tensor) else s.videos
+                    if isinstance(s_vids, torch.Tensor):
+                        videos.append(s_vids)
+                if len(s.video_thw_grids) > 0:
+                    if isinstance(s.video_thw_grids, np.ndarray):
+                        for row in s.video_thw_grids:
+                            video_thw_grids.append(row)
+                    else:
+                        video_thw_grids.extend(s.video_thw_grids)
+
+        return imgs, image_thw_grids, videos, video_thw_grids
+
+    def batch(self, samples: List[Union[QwenVLTaskSample, QwenVLTaskSamplePacked]]) -> QwenVLTaskBatch:
+        """
+        Put encoded sample into Batch, do padding, add labels and visual input masks.
+
+        Handles both unpacked (QwenVLTaskSample) and packed (QwenVLTaskSamplePacked) inputs.
+        For packed inputs, pads to self.seq_len and passes cu_lengths through.
+        """
+        is_packed = isinstance(samples[0], QwenVLTaskSamplePacked)
+
+        imgs, image_thw_grids, videos, video_thw_grids = self._collect_visual_data(samples)
+
+        pad_token_id = self.hf_tokenizer.pad_token_id
+
+        if is_packed:
+            batch_size = len(samples)
+            fixed_len = self.seq_len
+            low_content_threshold = _thd_diag_low_content_threshold()
+            max_low_content_logs = _thd_diag_max_low_content_logs()
+            diag_enabled = _thd_diag_enabled()
+
+            text_mat = np.full((batch_size, fixed_len), pad_token_id, dtype=np.int64)
+            target_mat = np.full((batch_size, fixed_len), pad_token_id, dtype=np.int64)
+            image_input_masks = np.zeros((batch_size, fixed_len), dtype=bool)
+            video_input_masks = np.zeros((batch_size, fixed_len), dtype=bool)
+
+            cu_lengths_list = []
+            max_lengths_list = []
+
+            for i, s in enumerate(samples):
+                content_len = len(s.tokens)
+                actual_len = min(content_len, fixed_len)
+                text_mat[i, :actual_len] = np.array(s.tokens)[:actual_len]
+                target_mat[i, :actual_len] = np.array(s.target)[:actual_len]
+                image_input_masks[i, :actual_len] = np.array(s.image_input_mask)[:actual_len]
+                video_input_masks[i, :actual_len] = np.array(s.video_input_mask)[:actual_len]
+
+                cu = s.cu_lengths.clone()
+                cu[cu > fixed_len] = fixed_len
+                cu_lengths_list.append(cu)
+                max_lengths_list.append(s.max_length)
+
+                pad_len = fixed_len - actual_len
+                if diag_enabled:
+                    logger.info(
+                        f"[BatchStats] sample {i}: {s.num_sub_samples} sub-seqs packed, "
+                        f"content={actual_len}, pad={pad_len}, "
+                        f"image_toks={s.num_image_tokens}, video_toks={s.num_video_tokens}, "
+                        f"text_toks={s.num_text_tokens}, vit_patches={s.num_vit_patches}, "
+                        f"utilization={actual_len / fixed_len * 100:.1f}%"
+                    )
+                    global _LOW_CONTENT_LOG_COUNT
+                    if (
+                        low_content_threshold > 0
+                        and actual_len < low_content_threshold
+                        and _LOW_CONTENT_LOG_COUNT < max_low_content_logs
+                    ):
+                        logger.warning(
+                            "[LowContentPackedSample] key=%s content=%d pad=%d seq_len=%d utilization=%.1f%% "
+                            "sub_samples=%d sub_lengths=%s image_toks=%d text_toks=%d",
+                            s.__key__,
+                            actual_len,
+                            pad_len,
+                            fixed_len,
+                            actual_len / fixed_len * 100.0,
+                            s.num_sub_samples,
+                            s.sub_sample_lengths,
+                            s.num_image_tokens,
+                            s.num_text_tokens,
+                        )
+                        _LOW_CONTENT_LOG_COUNT += 1
+
+            tokens = torch.from_numpy(text_mat)
+            tokens[tokens == pad_token_id] = 0
+            labels = torch.from_numpy(target_mat)
+            labels[labels == pad_token_id] = IGNORE_INDEX
+
+            attention_mask, loss_mask, position_ids = get_ltor_masks_and_position_ids(
+                data=tokens,
+                eod_token=self.hf_tokenizer.eos_token_id,
+                eod_mask_loss=False,
+                reset_attention_mask=False,
+                reset_position_ids=False,
+                compute_attention_mask=False,
+            )
+            loss_mask[labels < 0] = 0.0
+
+            max_cu_len = max(len(cu) for cu in cu_lengths_list)
+            padded_cu = torch.zeros((batch_size, max_cu_len), dtype=torch.int32)
+            for i, cu in enumerate(cu_lengths_list):
+                padded_cu[i, : len(cu)] = cu
+
+            cu_seqlens_argmin_list = [len(cu) for cu in cu_lengths_list]
+            cu_seqlens_argmin = torch.tensor(cu_seqlens_argmin_list, dtype=torch.int32)
+            max_lengths = torch.tensor(max_lengths_list, dtype=torch.int32)
+
+            sample_keys = [s.__key__ for s in samples]
+            batch_obj = QwenVLTaskBatch(
+                __key__="+".join(sample_keys),
+                __restore_key__=(),
+                __keys__=sample_keys,
+                __subflavors__=[s.__subflavors__ for s in samples],
+                pixel_values=self._concat_visual_blocks(imgs, "image"),
+                pixel_values_videos=self._concat_visual_blocks(videos, "video"),
+                image_grid_thw=torch.from_numpy(np.array(image_thw_grids)) if len(image_thw_grids) > 0 else None,
+                video_grid_thw=torch.from_numpy(np.array(video_thw_grids)) if len(video_thw_grids) > 0 else None,
+                image_input_mask=torch.from_numpy(image_input_masks),
+                video_input_mask=torch.from_numpy(video_input_masks),
+                input_ids=tokens,
+                attention_mask=None,
+                position_ids=position_ids,
+                labels=labels,
+                loss_mask=loss_mask,
+                cu_lengths=padded_cu,
+                max_lengths=max_lengths,
+                cu_seqlens_argmin=cu_seqlens_argmin,
+            )
+            return batch_obj
+
+        # --- Original unpacked path ---
         max_seq_len = max(len(s.text) for s in samples)
         if max_seq_len > self.seq_len:
             logging.warning("max sequence length larger than passed parameter")
 
-        text_mat = np.full((len(samples), max_seq_len), self.hf_tokenizer.pad_token_id, dtype=np.int64)
-        target_mat = np.full((len(samples), max_seq_len), self.hf_tokenizer.pad_token_id, dtype=np.int64)
+        text_mat = np.full((len(samples), max_seq_len), pad_token_id, dtype=np.int64)
+        target_mat = np.full((len(samples), max_seq_len), pad_token_id, dtype=np.int64)
 
         image_input_masks = np.zeros_like(text_mat, dtype=bool)
         video_input_masks = np.zeros_like(text_mat, dtype=bool)
         for i, s in enumerate(samples):
-            # If the sample/target length exceeds the target sequence length, then truncate.
             text_len = min(max_seq_len, len(s.text))
             target_len = min(max_seq_len, len(s.target))
 
             text_mat[i, :text_len] = np.array(s.text)[:text_len]
-            # NOTE: we should assert user input sequence will not be truncated
             if s.image_input_mask is not None:
                 image_input_masks[i, :text_len] = np.array(s.image_input_mask)[:text_len]
             if s.video_input_mask is not None:
@@ -395,10 +797,10 @@ def batch(self, samples: List[QwenVLTaskSample]) -> QwenVLTaskBatch:
             target_mat[i, :target_len] = np.array(s.target)[:target_len]
 
         tokens = torch.from_numpy(text_mat)
-        tokens[tokens == self.hf_tokenizer.pad_token_id] = 0
+        tokens[tokens == pad_token_id] = 0
 
         labels = torch.from_numpy(target_mat)
-        labels[labels == self.hf_tokenizer.pad_token_id] = IGNORE_INDEX
+        labels[labels == pad_token_id] = IGNORE_INDEX
 
         attention_mask, loss_mask, position_ids = get_ltor_masks_and_position_ids(
             data=tokens,
@@ -410,11 +812,14 @@ def batch(self, samples: List[QwenVLTaskSample]) -> QwenVLTaskBatch:
 
         loss_mask[labels < 0] = 0.0
 
-        batch = QwenVLTaskBatch(
-            __keys__=[s.__key__ for s in samples],
+        sample_keys = [s.__key__ for s in samples]
+        batch_obj = QwenVLTaskBatch(
+            __key__="+".join(sample_keys),
+            __restore_key__=(),
+            __keys__=sample_keys,
             __subflavors__=[s.__subflavors__ for s in samples],
-            pixel_values=torch.vstack(imgs) if len(imgs) > 0 else None,
-            pixel_values_videos=torch.vstack(videos) if len(videos) > 0 else None,
+            pixel_values=self._concat_visual_blocks(imgs, "image"),
+            pixel_values_videos=self._concat_visual_blocks(videos, "video"),
             image_grid_thw=torch.from_numpy(np.array(image_thw_grids)) if len(image_thw_grids) > 0 else None,
             video_grid_thw=torch.from_numpy(np.array(video_thw_grids)) if len(video_thw_grids) > 0 else None,
             image_input_mask=torch.from_numpy(image_input_masks),
@@ -425,7 +830,7 @@ def batch(self, samples: List[QwenVLTaskSample]) -> QwenVLTaskBatch:
             labels=labels,
             loss_mask=loss_mask,
         )
-        return batch
+        return batch_obj
 
     def encode_batch(self, batch: QwenVLTaskBatch) -> dict:
         """Encode batch in dict"""
@@ -438,4 +843,9 @@ def encode_batch(self, batch: QwenVLTaskBatch) -> dict:
             image_grid_thw=batch.image_grid_thw,
         )
 
+        if batch.cu_lengths is not None:
+            raw["cu_seqlens"] = batch.cu_lengths
+            raw["max_seqlen"] = batch.max_lengths
+            raw["cu_seqlens_argmin"] = batch.cu_seqlens_argmin
+
         return raw
diff --git a/src/megatron/bridge/recipes/qwen_vl/qwen35_vl.py b/src/megatron/bridge/recipes/qwen_vl/qwen35_vl.py
index af089cc28b..a90d908a18 100644
--- a/src/megatron/bridge/recipes/qwen_vl/qwen35_vl.py
+++ b/src/megatron/bridge/recipes/qwen_vl/qwen35_vl.py
@@ -28,7 +28,11 @@
 from megatron.bridge import AutoBridge
 from megatron.bridge.peft.base import PEFT
 from megatron.bridge.recipes.common import _peft_common_vlm, _sft_common_vlm
-from megatron.bridge.recipes.qwen_vl.qwen3_vl import Qwen3VLCommonKwargs, _qwen3_vl_common
+from megatron.bridge.recipes.qwen_vl.qwen3_vl import (
+    Qwen3VLCommonKwargs,
+    _make_energon_dataset,
+    _qwen3_vl_common,
+)
 from megatron.bridge.recipes.utils.finetune_utils import default_peft_config
 from megatron.bridge.recipes.utils.optimizer_utils import distributed_fused_adam_with_cosine_annealing
 from megatron.bridge.training.config import ConfigContainer
@@ -385,6 +389,52 @@ def qwen35_vl_35b_a3b_sft_config(hf_path: str = "Qwen/Qwen3.5-35B-A3B") -> Confi
     return cfg
 
 
+def qwen35_vl_35b_a3b_sft_energon_config(hf_path: str = "Qwen/Qwen3.5-35B-A3B") -> ConfigContainer:
+    """Return a full SFT config for Qwen3.5-VL 35B-A3B (MoE) with Energon dataset.
+
+    Same as qwen35_vl_35b_a3b_sft_config but uses EnergonProvider instead of HF dataset.
+    Packing is disabled by default. Enable THD batch-level packing via CLI:
+    dataset.batch_level_packing=true
+    Set the dataset path via CLI override: dataset.path=/path/to/energon/dataset
+    Optional fixed-bin overrides:
+      - dataset.cord_bins_root=/path/to/cord_bins
+      - dataset.cord_bin_prefix=cord_bin_
+      - dataset.cord_bin_id=0
+    """
+    cfg = qwen35_vl_35b_a3b_sft_config(hf_path=hf_path)
+    cfg.dataset = _make_energon_dataset(
+        hf_path, cfg.dataset.seq_length, cfg.train.micro_batch_size, cfg.train.global_batch_size
+    )
+    # Keep both packing modes disabled by default.
+    # - in-batch packing: dataset.pack_sequences_in_batch
+    # - THD batch-level packing: dataset.batch_level_packing
+    # Enable either mode explicitly via CLI override when needed.
+    cfg.dataset.pack_sequences_in_batch = False
+    cfg.dataset.batch_level_packing = False
+    # Default packing window; can still be overridden via CLI:
+    # dataset.packing_buffer_size=<N>
+    cfg.dataset.packing_buffer_size = 128
+    return cfg
+
+
+def qwen35_vl_35b_a3b_sft_energon_nopack_config(hf_path: str = "Qwen/Qwen3.5-35B-A3B") -> ConfigContainer:
+    """Return 35B-A3B Energon SFT config with packing disabled (BSHD baseline path).
+
+    This is useful as a non-THD reference when comparing with the THD recipe.
+    Set the dataset path via CLI override: dataset.path=/path/to/energon/dataset
+    Supports the same optional fixed-bin overrides as the THD Energon recipe.
+    """
+    cfg = qwen35_vl_35b_a3b_sft_config(hf_path=hf_path)
+    cfg.dataset = _make_energon_dataset(
+        hf_path, cfg.dataset.seq_length, cfg.train.micro_batch_size, cfg.train.global_batch_size
+    )
+    # Baseline path: keep Energon data flow but disable packing.
+    cfg.dataset.pack_sequences_in_batch = False
+    cfg.dataset.batch_level_packing = False
+    cfg.dataset.packing_buffer_size = None
+    return cfg
+
+
 def qwen35_vl_35b_a3b_fsdp_sft_config(hf_path: str = "Qwen/Qwen3.5-35B-A3B") -> ConfigContainer:
     """Return a full SFT config for Qwen3.5-VL 35B-A3B (MoE) with Megatron FSDP.
 
diff --git a/src/megatron/bridge/recipes/qwen_vl/qwen3_vl.py b/src/megatron/bridge/recipes/qwen_vl/qwen3_vl.py
index 5f84009620..ae6b428f99 100644
--- a/src/megatron/bridge/recipes/qwen_vl/qwen3_vl.py
+++ b/src/megatron/bridge/recipes/qwen_vl/qwen3_vl.py
@@ -19,19 +19,32 @@
 
 from __future__ import annotations
 
+import logging
 import os
 from typing import Optional, Union
 
 import torch
-from transformers import AutoTokenizer, Qwen3VLProcessor
+from transformers import AutoTokenizer
 from typing_extensions import TypedDict, Unpack
 
+
+logger = logging.getLogger(__name__)
+
+try:
+    from transformers import Qwen3VLProcessor
+except (ImportError, ModuleNotFoundError) as exc:
+    logger.warning(
+        "Qwen3VLProcessor import unavailable; falling back to AutoProcessor: %s",
+        exc,
+    )
+    Qwen3VLProcessor = None
+
 from megatron.bridge import AutoBridge
 from megatron.bridge.data.energon.energon_provider import EnergonProvider
 from megatron.bridge.data.vlm_datasets import MockVLMConversationProvider
 from megatron.bridge.peft.base import PEFT
 from megatron.bridge.recipes.common import _peft_common_vlm, _sft_common_vlm
-from megatron.bridge.recipes.qwen_vl.data.energon.task_encoder import QwenVLTaskEncoder
+from megatron.bridge.recipes.qwen_vl.data.energon.task_encoder import QwenVLTaskEncoder, _resolve_hf_mm_token_ids
 from megatron.bridge.recipes.utils.finetune_utils import default_peft_config
 from megatron.bridge.recipes.utils.optimizer_utils import distributed_fused_adam_with_cosine_annealing
 from megatron.bridge.recipes.utils.tokenizer_utils import DEFAULT_NULL_TOKENIZER_VOCAB_SIZE
@@ -269,14 +282,49 @@ def _make_energon_dataset(
     hf_path: str, seq_length: int, micro_batch_size: int, global_batch_size: int
 ) -> EnergonProvider:
     """Create an EnergonProvider dataset config for Qwen3-VL recipes."""
+    bridge = AutoBridge.from_hf_pretrained(hf_path)
+    model_cfg = bridge.to_megatron_provider(load_weights=False)
     tokenizer = AutoTokenizer.from_pretrained(hf_path)
-    # Use Qwen3VLProcessor to match the HF flow (which uses AutoProcessor).
-    # This processor accepts both images and videos kwargs.
-    image_processor = Qwen3VLProcessor.from_pretrained(hf_path)
+    # Prefer Qwen3VLProcessor on newer transformers; fall back to AutoProcessor
+    # for older container images that do not yet expose that symbol or lack
+    # the video_processing sub-module required by from_pretrained.
+    image_processor = None
+    if Qwen3VLProcessor is not None:
+        try:
+            image_processor = Qwen3VLProcessor.from_pretrained(hf_path)
+        except (AttributeError, OSError, RuntimeError, ValueError) as exc:
+            logger.warning(
+                "Qwen3VLProcessor.from_pretrained failed, falling back to AutoProcessor: %s",
+                exc,
+            )
+    if image_processor is None:
+        from transformers import AutoProcessor
+
+        image_processor = AutoProcessor.from_pretrained(hf_path)
+    image_token_id, video_token_id = _resolve_hf_mm_token_ids(tokenizer)
+    if image_token_id != model_cfg.image_token_id or video_token_id != model_cfg.video_token_id:
+        raise ValueError(
+            "Tokenizer multimodal token ids do not match model config: "
+            f"tokenizer(image={image_token_id}, video={video_token_id}) vs "
+            f"model(image={model_cfg.image_token_id}, video={model_cfg.video_token_id})"
+        )
+
+    min_pixels = getattr(image_processor, "min_pixels", None)
+    max_pixels = getattr(image_processor, "max_pixels", None)
+    if min_pixels is None or max_pixels is None:
+        raise ValueError("Processor must expose both min_pixels and max_pixels for strict task encoder config")
+
     task_encoder = QwenVLTaskEncoder(
         tokenizer=tokenizer,
         image_processor=image_processor,
+        temporal_patch_size=model_cfg.temporal_patch_size,
+        spatial_merge_size=model_cfg.spatial_merge_size,
+        patch_size=model_cfg.patch_size,
+        image_token_id=model_cfg.image_token_id,
+        video_token_id=model_cfg.video_token_id,
         max_padding_length=seq_length,
+        min_pixels=int(min_pixels),
+        max_pixels=int(max_pixels),
     )
     return EnergonProvider(
         path="",  # Must be set via CLI override: dataset.path=<path>
diff --git a/src/megatron/bridge/training/config.py b/src/megatron/bridge/training/config.py
index 6217f282c0..e5c06d4da5 100644
--- a/src/megatron/bridge/training/config.py
+++ b/src/megatron/bridge/training/config.py
@@ -1129,9 +1129,20 @@ def validate(self) -> None:
         Ensures compatibility between different configuration settings.
         """
 
-        # Propagate in-batch packing flag to model config so TransformerConfig.finalize()
+        in_batch_pack_enabled = getattr(self.dataset, "pack_sequences_in_batch", False)
+        batch_level_pack_enabled = getattr(self.dataset, "batch_level_packing", False)
+
+        # Keep in-batch packing and batch-level packing as two independent modes.
+        # They serve different semantics and should not be enabled together.
+        if in_batch_pack_enabled and batch_level_pack_enabled:
+            raise ValueError(
+                "dataset.pack_sequences_in_batch and dataset.batch_level_packing "
+                "cannot both be true. Keep in-batch packing and batch-level packing separate."
+            )
+
+        # Propagate packed-batch flags to model config so TransformerConfig.finalize()
         # can enable variable_seq_lengths for pipeline parallelism.
-        if getattr(self.dataset, "pack_sequences_in_batch", False):
+        if in_batch_pack_enabled or batch_level_pack_enabled:
             self.model._pack_sequences_in_batch = True
 
         if hasattr(self.dataset, "finalize"):
@@ -1305,10 +1316,11 @@ def validate(self) -> None:
                 f"https://docs.nvidia.com/nemo-framework/user-guide/latest/sft_peft/packed_sequence.html"
             )
 
-        if getattr(self.dataset, "pack_sequences_in_batch", False) and self.train.micro_batch_size == 1:
+        if in_batch_pack_enabled and self.train.micro_batch_size == 1:
             raise ValueError(
-                "micro_batch_size should be greater than 1 when using pack_sequences_in_batch=True. "
-                "In-batch packing concatenates multiple sequences within a microbatch, so at least 2 sequences "
+                "micro_batch_size should be greater than 1 when using in-batch packing "
+                "(dataset.pack_sequences_in_batch=True). In-batch packing concatenates multiple sequences "
+                "within a microbatch, so at least 2 sequences "
                 "are required per micro-batch."
             )
 
diff --git a/src/megatron/bridge/training/utils/packed_seq_utils.py b/src/megatron/bridge/training/utils/packed_seq_utils.py
index 9683a17278..2363c22e4a 100644
--- a/src/megatron/bridge/training/utils/packed_seq_utils.py
+++ b/src/megatron/bridge/training/utils/packed_seq_utils.py
@@ -18,12 +18,25 @@
 from megatron.core.packed_seq_params import PackedSeqParams
 
 
+def _trim_cu_seqlens(cu_seqlens: torch.Tensor, cu_seqlens_argmin: torch.Tensor | None) -> torch.Tensor:
+    """Trim padded cu_seqlens tail safely.
+
+    If argmin metadata is available from dataloader, trust it. Otherwise, detect
+    the first zero in the tail region after index 0 (the leading 0 is valid).
+    """
+    if cu_seqlens_argmin is not None:
+        return cu_seqlens[: int(cu_seqlens_argmin.item())]
+
+    padded_tail = torch.nonzero(cu_seqlens[1:] == 0, as_tuple=False)
+    first_tail = int(padded_tail[0].item() + 1) if padded_tail.numel() else cu_seqlens.numel()
+    return cu_seqlens[:first_tail]
+
+
 def get_packed_seq_params(batch: dict[str, torch.Tensor]) -> PackedSeqParams:
     """Build packed sequence parameters from a batch dictionary.
 
-    The function squeezes possible batch dimensions and removes any padding
-    marked by -1 values. It returns a `PackedSeqParams` instance suitable for
-    packed sequence attention kernels.
+    The function trims optional sentinel tails via argmin metadata and returns
+    a `PackedSeqParams` instance suitable for packed sequence kernels.
 
     Args:
         batch: A dictionary containing packed-sequence metadata. Expected keys:
@@ -32,8 +45,9 @@ def get_packed_seq_params(batch: dict[str, torch.Tensor]) -> PackedSeqParams:
             hybrid SSM/Mamba models to generate ``seq_idx``).
 
     Returns:
-        PackedSeqParams with identical q/kv parameters and `qkv_format` set to
-        "thd".
+        PackedSeqParams for THD (`qkv_format="thd"`). If unpadded boundaries
+        are provided, they are used for q/kv while padded boundaries remain in
+        `*_padded` fields for kernel compatibility.
     """
 
     cu_seqlens_padded = batch["cu_seqlens"].squeeze()
@@ -44,18 +58,12 @@ def get_packed_seq_params(batch: dict[str, torch.Tensor]) -> PackedSeqParams:
     cu_seqlens_argmin = batch.get("cu_seqlens_argmin")
     cu_seqlens_unpadded_argmin = batch.get("cu_seqlens_unpadded_argmin")
 
-    # note: if argmin is not pre-computed in the dataloader, torch.argmin here will incur a
-    # device-to-host synchronization, which can slow down training
-    if cu_seqlens_argmin is not None:
-        cu_seqlens_padded = cu_seqlens_padded[: cu_seqlens_argmin.item()]
-    else:
-        cu_seqlens_padded = cu_seqlens_padded[: torch.argmin(cu_seqlens_padded)]
+    # Note: if argmin metadata is absent, fallback tail detection can still incur
+    # device-to-host synchronization.
+    cu_seqlens_padded = _trim_cu_seqlens(cu_seqlens_padded, cu_seqlens_argmin)
 
     if cu_seqlens_unpadded is not None:
-        if cu_seqlens_unpadded_argmin is not None:
-            cu_seqlens_unpadded = cu_seqlens_unpadded[: cu_seqlens_unpadded_argmin.item()]
-        else:
-            cu_seqlens_unpadded = cu_seqlens_unpadded[: torch.argmin(cu_seqlens_unpadded)]
+        cu_seqlens_unpadded = _trim_cu_seqlens(cu_seqlens_unpadded, cu_seqlens_unpadded_argmin)
 
     max_seqlen = batch["max_seqlen"].squeeze() if "max_seqlen" in batch else None
     total_tokens = batch.get("total_tokens")
@@ -74,9 +82,14 @@ def get_packed_seq_params(batch: dict[str, torch.Tensor]) -> PackedSeqParams:
             qkv_format="thd",
         )
     else:
+        # Follow Megatron-LM data_schedule.py convention: set all four
+        # cu_seqlens fields to padded values so packed kernels consume a
+        # single consistent boundary set.
         return PackedSeqParams(
             cu_seqlens_q=cu_seqlens_padded,
             cu_seqlens_kv=cu_seqlens_padded,
+            cu_seqlens_q_padded=cu_seqlens_padded,
+            cu_seqlens_kv_padded=cu_seqlens_padded,
             max_seqlen_q=max_seqlen,
             max_seqlen_kv=max_seqlen,
             total_tokens=total_tokens,
diff --git a/src/megatron/bridge/training/vlm_step.py b/src/megatron/bridge/training/vlm_step.py
index 3137cc36df..4b0b67e249 100644
--- a/src/megatron/bridge/training/vlm_step.py
+++ b/src/megatron/bridge/training/vlm_step.py
@@ -14,6 +14,7 @@
 
 import logging
 import math
+import os
 from functools import partial
 from typing import Any, Iterable
 
@@ -34,9 +35,50 @@
     pad_or_truncate_pos_to_len,
 )
 from megatron.bridge.training.utils.pg_utils import get_pg_collection
+from megatron.bridge.utils.common_utils import is_rank_0
 
 
 logger = logging.getLogger(__name__)
+_SWITCH_LOGGED: set[str] = set()
+
+
+def _thd_diag_enabled() -> bool:
+    return _switch_enabled("THD_DIAG")
+
+
+def _thd_diag_boundary_enabled() -> bool:
+    return _switch_enabled("THD_DIAG_BOUNDARY")
+
+
+def _switch_enabled(name: str) -> bool:
+    enabled = os.environ.get(name, "0") not in ("0", "", "false", "False")
+    if enabled and is_rank_0() and name not in _SWITCH_LOGGED:
+        logger.info("[THD_SWITCH] %s=1 enabled", name)
+        _SWITCH_LOGGED.add(name)
+    return enabled
+
+
+def _thd_force_bshd_enabled() -> bool:
+    return _switch_enabled("THD_FORCE_BSHD")
+
+
+def _thd_force_single_segment_cu_enabled() -> bool:
+    return _switch_enabled("THD_FORCE_SINGLE_SEGMENT_CU")
+
+
+def _thd_skip_preprocess_packed_pos_enabled() -> bool:
+    return _switch_enabled("THD_SKIP_PREPROCESS_PACKED_POS")
+
+
+def _resolve_cfg_token_id(cfg: ConfigContainer, key: str) -> int | None:
+    model_cfg = getattr(cfg, "model", None)
+    token_id = getattr(model_cfg, key, None)
+    if token_id is None:
+        return None
+    try:
+        return int(token_id)
+    except (TypeError, ValueError):
+        return None
 
 
 def get_batch_from_iterator(
@@ -70,8 +112,12 @@ def get_batch_from_iterator(
 
     if "cu_seqlens" in batch:
         required_device_keys.add("cu_seqlens")
+        if "cu_seqlens_unpadded" in batch:
+            required_device_keys.add("cu_seqlens_unpadded")
         required_host_keys.add("cu_seqlens_argmin")
         required_host_keys.add("max_seqlen")
+        if "cu_seqlens_unpadded_argmin" in batch:
+            required_host_keys.add("cu_seqlens_unpadded_argmin")
 
     required_device_keys.update(("tokens", "input_ids", "position_ids"))
     if is_last_pp_stage:
@@ -192,7 +238,7 @@ def pack_batch_sequences(
         else:
             padded_len = length
         padded_seq_lengths.append(padded_len)
-        # Use padded lengths for cu_seqlens so THD RoPE splits sum correctly under CP.
+        # Use padded lengths for packed kernels under CP/SP alignment constraints.
         cu_seqlens.append(cu_seqlens[-1] + padded_len)
 
     cu_seqlens = torch.tensor(cu_seqlens, dtype=torch.int32, device=device)
@@ -233,9 +279,10 @@ def pack_batch_sequences(
             )
         offset += padded_len
 
-    logger.debug(
-        f"Packed {len(valid_sequences)} sequences: lengths={seq_lengths}, total_len={total_len}, max_len={max_seqlen}"
-    )
+    if _thd_diag_enabled():
+        logger.debug(
+            f"Packed {len(valid_sequences)} sequences: lengths={seq_lengths}, total_len={total_len}, max_len={max_seqlen}"
+        )
 
     # Attention mask is not used with packed sequences (handled by cu_seqlens)
     packed_attention_mask = None
@@ -275,8 +322,16 @@ def get_batch(data_iterator: Iterable, cfg: ConfigContainer, use_mtp: bool = Fal
         is_first_pp_stage=is_first,
         is_last_pp_stage=is_last,
     )
-    enable_packing = getattr(cfg.dataset, "pack_sequences_in_batch", False)
-
+    in_batch_pack_enabled = getattr(cfg.dataset, "pack_sequences_in_batch", False)
+    batch_level_pack_enabled = getattr(cfg.dataset, "batch_level_packing", False)
+    enable_packing = in_batch_pack_enabled or batch_level_pack_enabled
+    force_bshd = _thd_force_bshd_enabled()
+    force_single_segment_cu = _thd_force_single_segment_cu_enabled()
+
+    if force_bshd and force_single_segment_cu and is_rank_0():
+        logger.warning(
+            "[THD_SWITCH] THD_FORCE_BSHD=1 overrides THD_FORCE_SINGLE_SEGMENT_CU=1 (single-segment CU ignored)."
+        )
     if not enable_packing:
         # When using pipeline parallelism, ensure fixed shapes equal to cfg.model.seq_length
         if getattr(cfg.model, "pipeline_model_parallel_size", 1) > 1:
@@ -335,8 +390,71 @@ def _ceil_to_mult(n: int, mult: int) -> int:
     tp_size = pg_collection.tp.size() if pg_collection is not None and pg_collection.tp is not None else 1
     has_sp = getattr(cfg.model, "sequence_parallel", False)
 
+    if force_bshd:
+        # Explicitly disable packed paths and ignore pre-packed cu_seqlens metadata.
+        enable_packing = False
+
+    # Energon pre-packed path: cu_seqlens already present from TaskEncoder packing
+    energon_cu_seqlens = batch.get("cu_seqlens")
+    if energon_cu_seqlens is not None and not force_bshd:
+        tokens_or_input = batch.get("tokens") if batch.get("tokens") is not None else batch.get("input_ids")
+        energon_max_seqlen = batch.get("max_seqlen")
+        energon_cu_argmin = batch.get("cu_seqlens_argmin")
+
+        if force_single_segment_cu and tokens_or_input is not None:
+            seq_len = int(tokens_or_input.shape[-1])
+            device = (
+                energon_cu_seqlens.device if isinstance(energon_cu_seqlens, torch.Tensor) else tokens_or_input.device
+            )
+            energon_cu_seqlens = torch.tensor([0, seq_len], dtype=torch.int32, device=device)
+            energon_max_seqlen = torch.tensor(seq_len, dtype=torch.int32, device=device)
+            energon_cu_argmin = torch.tensor(2, dtype=torch.int64)
+            if _thd_diag_enabled() and is_rank_0():
+                logger.info("[THD_SWITCH] force single-segment cu_seqlens=[0, %d]", seq_len)
+
+        # Log detailed packed-iteration diagnostics when explicitly enabled.
+        if tokens_or_input is not None and _thd_diag_enabled():
+            _seq_dim = tokens_or_input.shape[-1]
+            if energon_cu_seqlens.dim() == 1:
+                _content_len = energon_cu_seqlens[-1].item()
+            elif energon_cu_argmin is not None:
+                _argmin_val = energon_cu_argmin.item() if energon_cu_argmin.dim() == 0 else energon_cu_argmin[0].item()
+                _content_len = energon_cu_seqlens[0, _argmin_val - 1].item()
+            else:
+                _content_len = energon_cu_seqlens[0, -1].item()
+            _pad_len = _seq_dim - _content_len
+            image_token_id = _resolve_cfg_token_id(cfg, "image_token_id")
+            if image_token_id is not None:
+                _n_img_toks: int | str = int((tokens_or_input == image_token_id).sum().item())
+            else:
+                _n_img_toks = "n/a"
+            _vit_shape = (
+                visual_inputs.pixel_values.shape
+                if visual_inputs is not None and visual_inputs.pixel_values is not None
+                else "None"
+            )
+            logger.info(
+                f"[IterStats] decoder_input={list(tokens_or_input.shape)}, "
+                f"content={_content_len}, pad={_pad_len}, "
+                f"image_tokens_in_decoder={_n_img_toks}, "
+                f"image_token_id={image_token_id}, "
+                f"vit_input={_vit_shape}"
+            )
+
+        return (
+            tokens_or_input,
+            batch.get("labels"),
+            batch.get("loss_mask"),
+            batch.get("attention_mask"),
+            batch.get("position_ids"),
+            energon_cu_seqlens,
+            energon_max_seqlen,
+            visual_inputs,
+            energon_cu_argmin,
+        )
+
     if enable_packing:
-        # Pack sequences
+        # In-batch packing (concatenate all micro-batch sequences)
         tokens_or_input = batch.get("tokens") if batch.get("tokens") is not None else batch.get("input_ids")
 
         # Compute pad_to_multiple_of as lcm of CP and SP constraints.
@@ -366,7 +484,6 @@ def _ceil_to_mult(n: int, mult: int) -> int:
             padding_mask=batch.get("_padding_mask"),
         )
 
-        # Update batch dict with packed tensors
         if batch.get("tokens") is not None:
             batch["tokens"] = packed_tokens
         else:
@@ -376,10 +493,9 @@ def _ceil_to_mult(n: int, mult: int) -> int:
         batch["attention_mask"] = packed_attention_mask
         batch["position_ids"] = packed_position_ids
 
-        # # Add packing metadata
-        logger.debug(f"Packed batch: cu_seqlens={cu_seqlens.tolist()}, max_seqlen={max_seqlen}")
+        if _thd_diag_enabled():
+            logger.debug(f"In-batch packed: cu_seqlens={cu_seqlens.tolist()}, max_seqlen={max_seqlen}")
     else:
-        # No packing, use dummy values
         cu_seqlens = None
         max_seqlen = None
 
@@ -392,6 +508,7 @@ def _ceil_to_mult(n: int, mult: int) -> int:
         cu_seqlens,
         max_seqlen,
         visual_inputs,
+        None,  # cu_seqlens_argmin (only set for Energon pre-packed path)
     )
 
 
@@ -427,12 +544,17 @@ def forward_step(
             cu_seqlens,
             max_seqlen,
             visual_inputs,
+            cu_seqlens_argmin,
         ) = get_batch(data_iterator, state.cfg, use_mtp, pg_collection=pg_collection)
     timers("batch-generator").stop()
 
+    # For Qwen-VL, let model compute MRoPE position_ids from multimodal tokens.
+    # Passing 1D sequential position_ids from dataloader would bypass that path.
+    model_position_ids = None if visual_inputs is not None else position_ids
+
     forward_args = {
         "input_ids": tokens,
-        "position_ids": position_ids,
+        "position_ids": model_position_ids,
         "attention_mask": attention_mask,
         "labels": labels,
         "loss_mask": loss_mask,  # Pass full loss_mask so model can slice it consistently with labels
@@ -443,18 +565,126 @@ def forward_step(
 
     # Add packed sequence support
     if cu_seqlens is not None:
-        cu_seqlens_argmin = torch.tensor(len(cu_seqlens))  # no padding in cu_seqlens since packing is done in-batch
-        packed_seq_params = {
-            "cu_seqlens": cu_seqlens,
-            "max_seqlen": max_seqlen,
-            "cu_seqlens_argmin": cu_seqlens_argmin,
+        physical_seq_len = tokens.shape[-1]
+
+        # Determine the number of valid entries in cu_seqlens.
+        cu = cu_seqlens.squeeze()
+        if cu_seqlens_argmin is not None:
+            n_valid = int(cu_seqlens_argmin.item())
+        elif cu.dim() == 1:
+            n_valid = len(cu)
+        else:
+            n_valid = int(torch.argmin(cu).item())
+
+        cu_clean = cu[:n_valid]
+        last_boundary = int(cu_clean[-1].item())
+
+        # Keep an unpadded copy for Qwen MRoPE sub-sequence semantics.
+        # Do not feed unpadded cu_seqlens into PackedSeqParams, which can
+        # switch TE kernels and destabilize training.
+        cu_unpadded = cu_clean
+
+        if last_boundary < physical_seq_len and not _thd_skip_preprocess_packed_pos_enabled():
+            # Trailing padding exists: extend padded boundary to full physical
+            # sequence length so packed kernels cover all tokens.
+            cu_padded = cu_clean.clone()
+            cu_padded[-1] = physical_seq_len
+            max_seqlen_out = (cu_padded[1:] - cu_padded[:-1]).max()
+        else:
+            if last_boundary < physical_seq_len and _thd_diag_enabled() and is_rank_0():
+                logger.info(
+                    "[THD_SWITCH] skip packed-pos preprocess: preserve original cu_seqlens last_boundary=%d physical_seq_len=%d",
+                    last_boundary,
+                    physical_seq_len,
+                )
+            cu_padded = cu_clean
+            if max_seqlen is not None:
+                max_seqlen_out = max_seqlen.squeeze()
+            else:
+                max_seqlen_out = (cu_padded[1:] - cu_padded[:-1]).max()
+
+        packed_seq_dict = {
+            "cu_seqlens": cu_padded,
+            "cu_seqlens_argmin": torch.tensor(len(cu_padded)),
+            "max_seqlen": max_seqlen_out,
         }
         # total_tokens drives seq_idx computation in PackedSeqParams.__post_init__,
         # which is only needed for Mamba/hybrid SSM layers. Skip it for pure
         # transformer models to avoid per-step CUDA overhead.
         if getattr(config, "is_hybrid_model", False):
-            packed_seq_params["total_tokens"] = tokens.size(1) if tokens is not None else labels.size(1)
-        forward_args["packed_seq_params"] = get_packed_seq_params(packed_seq_params)
+            packed_seq_dict["total_tokens"] = tokens.size(1) if tokens is not None else labels.size(1)
+        # Build an explicit padding mask for MoE routing/aux-loss accounting.
+        # True means padding token that should be excluded from MoE statistics.
+        moe_padding_mask = torch.zeros_like(tokens, dtype=torch.bool)
+        if cu_unpadded.numel() == cu_padded.numel() and cu_padded.numel() > 1:
+            for i in range(int(cu_padded.numel()) - 1):
+                seg_unpadded_end = int(cu_unpadded[i + 1].item())
+                seg_padded_end = int(cu_padded[i + 1].item())
+                if seg_padded_end > seg_unpadded_end:
+                    moe_padding_mask[..., seg_unpadded_end:seg_padded_end] = True
+
+        forward_args["packed_seq_params"] = get_packed_seq_params(packed_seq_dict)
+        # Pass unpadded boundaries only to Qwen model's MRoPE construction.
+        forward_args["rope_cu_seqlens"] = cu_unpadded
+        forward_args["moe_padding_mask"] = moe_padding_mask
+        if _thd_diag_enabled() and is_rank_0():
+            logger.info(
+                "[THD_DIAG][packed] physical_seq_len=%d cu_unpadded_last=%d cu_padded_last=%d implicit_pad=%d max_seqlen=%d moe_padding_tokens=%d",
+                int(physical_seq_len),
+                int(cu_unpadded[-1].item()),
+                int(cu_padded[-1].item()),
+                int(physical_seq_len - int(cu_unpadded[-1].item())),
+                int(max_seqlen_out.item()) if torch.is_tensor(max_seqlen_out) else int(max_seqlen_out),
+                int(moe_padding_mask.sum().item()),
+            )
+        if _thd_diag_boundary_enabled() and is_rank_0():
+            cu_padded_cpu = cu_padded.detach().cpu()
+            cu_unpadded_cpu = cu_unpadded.detach().cpu()
+            seg_lens_padded = (cu_padded_cpu[1:] - cu_padded_cpu[:-1]).tolist()
+            seg_lens_unpadded = (cu_unpadded_cpu[1:] - cu_unpadded_cpu[:-1]).tolist()
+            logger.info(
+                "[THD_DIAG][cu] kernel_num_segs=%d rope_num_segs=%d kernel_cu_head=%s rope_cu_head=%s kernel_seg_lens_head=%s rope_seg_lens_head=%s",
+                max(0, int(cu_padded_cpu.numel()) - 1),
+                max(0, int(cu_unpadded_cpu.numel()) - 1),
+                cu_padded_cpu[:8].tolist(),
+                cu_unpadded_cpu[:8].tolist(),
+                seg_lens_padded[:8],
+                seg_lens_unpadded[:8],
+            )
+            # Boundary windows are centered on each segment join index.
+            # This helps detect edge-only mismatch that can be hidden by global stats.
+            if labels is not None and loss_mask is not None and cu_unpadded_cpu.numel() > 2:
+                win_k = 8
+                seq_len = int(tokens.shape[-1])
+                boundary_window = torch.zeros(seq_len, dtype=torch.bool, device=tokens.device)
+                boundary_positions = []
+                for i in range(1, int(cu_unpadded_cpu.numel()) - 1):
+                    join_idx = int(cu_unpadded_cpu[i].item())
+                    boundary_positions.append(join_idx)
+                    left = max(0, join_idx - win_k)
+                    right = min(seq_len, join_idx + win_k)
+                    if right > left:
+                        boundary_window[left:right] = True
+                labels_flat = labels.view(-1)
+                loss_mask_flat = loss_mask.view(-1)
+                moe_mask_flat = moe_padding_mask.view(-1)
+                bw = boundary_window.view(-1)
+                bw_tokens = int(bw.sum().item())
+                bw_valid_loss = int(((loss_mask_flat > 0) & bw).sum().item())
+                bw_valid_labels = int(((labels_flat != -100) & bw).sum().item())
+                bw_moe_padding = int((moe_mask_flat & bw).sum().item())
+                bw_label_loss_mismatch = int((((labels_flat != -100) != (loss_mask_flat > 0)) & bw).sum().item())
+                logger.info(
+                    "[THD_DIAG][boundary] win_k=%d joins=%d join_head=%s window_tokens=%d valid_loss=%d valid_labels=%d moe_padding=%d label_loss_mismatch=%d",
+                    win_k,
+                    len(boundary_positions),
+                    boundary_positions[:8],
+                    bw_tokens,
+                    bw_valid_loss,
+                    bw_valid_labels,
+                    bw_moe_padding,
+                    bw_label_loss_mismatch,
+                )
 
     if loss_mask is not None:
         loss_mask = loss_mask.contiguous()
@@ -479,6 +709,28 @@ def forward_step(
             else:
                 output_tensor = model_output
 
+    if _thd_diag_enabled() and is_rank_0() and loss_mask is not None:
+        loss_mask_flat = loss_mask.view(-1)
+        valid_loss_tokens = int((loss_mask_flat > 0).sum().item())
+        total_loss_tokens = int(loss_mask_flat.numel())
+        if labels is not None:
+            labels_flat = labels.view(-1)
+            ignore_tokens = int((labels_flat == -100).sum().item())
+            logger.info(
+                "[THD_DIAG][loss] total=%d valid=%d masked=%d labels_ignore=%d",
+                total_loss_tokens,
+                valid_loss_tokens,
+                total_loss_tokens - valid_loss_tokens,
+                ignore_tokens,
+            )
+        else:
+            logger.info(
+                "[THD_DIAG][loss] total=%d valid=%d masked=%d labels_ignore=n/a",
+                total_loss_tokens,
+                valid_loss_tokens,
+                total_loss_tokens - valid_loss_tokens,
+            )
+
     loss_function = _create_loss_function(loss_mask, check_for_nan_in_loss, check_for_spiky_loss)
 
     return output_tensor, loss_function
diff --git a/src/megatron/bridge/utils/common_utils.py b/src/megatron/bridge/utils/common_utils.py
index 35a3d507bd..0357a964cc 100644
--- a/src/megatron/bridge/utils/common_utils.py
+++ b/src/megatron/bridge/utils/common_utils.py
@@ -66,6 +66,11 @@ def get_rank_safe() -> int:
     return 0
 
 
+def is_rank_0() -> bool:
+    """Return whether the current process is rank 0."""
+    return get_rank_safe() == 0
+
+
 def get_world_size_safe() -> int:
     """Get the distributed world size safely, even if torch.distributed is not initialized.
 
diff --git a/tests/unit_tests/models/qwen_vl/test_diag_utils.py b/tests/unit_tests/models/qwen_vl/test_diag_utils.py
new file mode 100644
index 0000000000..17a13f4379
--- /dev/null
+++ b/tests/unit_tests/models/qwen_vl/test_diag_utils.py
@@ -0,0 +1,48 @@
+# Copyright (c) 2025, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest.mock import patch
+
+from megatron.bridge.models.qwen_vl.modelling_qwen3_vl.utils import (
+    thd_diag_align_enabled,
+    thd_diag_enabled,
+    thd_diag_mrope_enabled,
+)
+
+
+class TestDiagSwitches:
+    @patch.dict(os.environ, {"THD_DIAG": "1"}, clear=False)
+    def test_thd_diag_enabled_true(self):
+        assert thd_diag_enabled() is True
+
+    @patch.dict(os.environ, {"THD_DIAG": "false"}, clear=False)
+    def test_thd_diag_enabled_false(self):
+        assert thd_diag_enabled() is False
+
+    @patch.dict(os.environ, {"THD_DIAG_ALIGN": "1"}, clear=False)
+    def test_thd_diag_align_enabled_true(self):
+        assert thd_diag_align_enabled() is True
+
+    @patch.dict(os.environ, {"THD_DIAG_ALIGN": "False"}, clear=False)
+    def test_thd_diag_align_enabled_false(self):
+        assert thd_diag_align_enabled() is False
+
+    @patch.dict(os.environ, {"THD_DIAG_MROPE": "1"}, clear=False)
+    def test_thd_diag_mrope_enabled_true(self):
+        assert thd_diag_mrope_enabled() is True
+
+    @patch.dict(os.environ, {"THD_DIAG_MROPE": ""}, clear=False)
+    def test_thd_diag_mrope_enabled_false(self):
+        assert thd_diag_mrope_enabled() is False
diff --git a/tests/unit_tests/recipes/qwen_vl/data/energon/test_task_encoder.py b/tests/unit_tests/recipes/qwen_vl/data/energon/test_task_encoder.py
index bb4e83abbd..163949c963 100644
--- a/tests/unit_tests/recipes/qwen_vl/data/energon/test_task_encoder.py
+++ b/tests/unit_tests/recipes/qwen_vl/data/energon/test_task_encoder.py
@@ -32,6 +32,7 @@
     convert_to_qwenvl_content,
     find_pattern_indices,
     get_ltor_masks_and_position_ids,
+    greedy_knapsack,
     process_vision,
     videohandler,
 )
@@ -42,6 +43,28 @@ def cleanup_local_folder():
     pass
 
 
+@pytest.fixture
+def qwenvl_encoder_config():
+    """Centralized encoder config for unit tests."""
+    return {
+        "temporal_patch_size": 2,
+        "patch_size": 14,
+        "spatial_merge_size": 2,
+        "image_token_id": 151655,
+        "video_token_id": 151656,
+        "max_padding_length": 128,
+        "min_pixels": 200704,
+        "max_pixels": 1003520,
+    }
+
+
+@pytest.fixture(autouse=True)
+def inject_qwenvl_encoder_config(request, qwenvl_encoder_config):
+    """Inject shared encoder config into unittest-style test cases."""
+    if request.instance is not None:
+        request.instance.qwenvl_encoder_config = qwenvl_encoder_config
+
+
 class TestHelperFunctions(unittest.TestCase):
     def test_find_pattern_indices(self):
         seq = np.array([1, 2, 3, 4, 5])
@@ -87,6 +110,24 @@ def test_get_ltor_masks_and_position_ids(self):
         self.assertTrue(torch.all(loss_mask == 1.0))
 
 
+class TestGreedyKnapsack(unittest.TestCase):
+    def test_empty_input(self):
+        self.assertEqual(greedy_knapsack([], [], max_capacity=8), [])
+
+    def test_single_item(self):
+        bins = greedy_knapsack([4], ["a"], max_capacity=8)
+        self.assertEqual(bins, [["a"]])
+
+    def test_exact_fill(self):
+        bins = greedy_knapsack([5, 3], ["a", "b"], max_capacity=8)
+        self.assertEqual(len(bins), 1)
+        self.assertCountEqual(bins[0], ["a", "b"])
+
+    def test_item_exceeds_capacity(self):
+        with self.assertRaises(ValueError):
+            greedy_knapsack([9], ["a"], max_capacity=8)
+
+
 class TestResolveHfMmTokenIds(unittest.TestCase):
     def test_resolves_from_tokenizer_attributes(self):
         tokenizer = MagicMock()
@@ -105,14 +146,13 @@ def test_falls_back_to_convert_tokens_to_ids(self):
         self.assertEqual(image_id, 300)
         self.assertEqual(video_id, 400)
 
-    def test_returns_defaults_when_all_fail(self):
+    def test_raises_when_all_resolution_paths_fail(self):
         tokenizer = MagicMock()
         tokenizer.image_token_id = None
         tokenizer.video_token_id = None
         tokenizer.convert_tokens_to_ids.side_effect = Exception("not found")
-        image_id, video_id = _resolve_hf_mm_token_ids(tokenizer)
-        self.assertEqual(image_id, 151655)
-        self.assertEqual(video_id, 151656)
+        with self.assertRaises(ValueError):
+            _resolve_hf_mm_token_ids(tokenizer)
 
 
 class TestVideoHandler(unittest.TestCase):
@@ -164,9 +204,7 @@ def setUp(self):
         self.encoder = QwenVLTaskEncoder(
             tokenizer=self.tokenizer,
             image_processor=self.image_processor,
-            max_padding_length=128,
-            patch_size=14,
-            spatial_merge_size=2,
+            **self.qwenvl_encoder_config,
         )
 
     def test_process_vision(self):
diff --git a/tests/unit_tests/utils/test_common_utils.py b/tests/unit_tests/utils/test_common_utils.py
index e75ff20ba8..6ff4b076a8 100644
--- a/tests/unit_tests/utils/test_common_utils.py
+++ b/tests/unit_tests/utils/test_common_utils.py
@@ -27,6 +27,7 @@
     get_rank_safe,
     get_world_size_safe,
     is_last_rank,
+    is_rank_0,
     print_rank_0,
     print_rank_last,
 )
@@ -178,6 +179,24 @@ def test_no_print_on_non_zero_rank(self, mock_print, mock_get_rank_safe):
         mock_print.assert_not_called()
 
 
+class TestIsRank0:
+    """Test is_rank_0 function."""
+
+    @patch("megatron.bridge.utils.common_utils.get_rank_safe")
+    def test_rank_zero(self, mock_get_rank_safe):
+        """Test is_rank_0 returns True when rank is 0."""
+        mock_get_rank_safe.return_value = 0
+        assert is_rank_0() is True
+        mock_get_rank_safe.assert_called_once()
+
+    @patch("megatron.bridge.utils.common_utils.get_rank_safe")
+    def test_non_zero_rank(self, mock_get_rank_safe):
+        """Test is_rank_0 returns False when rank is non-zero."""
+        mock_get_rank_safe.return_value = 3
+        assert is_rank_0() is False
+        mock_get_rank_safe.assert_called_once()
+
+
 class TestIsLastRank:
     """Test is_last_rank function."""