diff --git a/megatron/training/arguments.py b/megatron/training/arguments.py
index d4f9eb9c0de..d519bc886c2 100644
--- a/megatron/training/arguments.py
+++ b/megatron/training/arguments.py
@@ -6,15 +6,15 @@
 import dataclasses
 import json
 import os
-from pathlib import Path
 import re
 import types
+from pathlib import Path
 
 import torch
 
+from megatron.core.msc_utils import MultiStorageClientFeature
 from megatron.core.rerun_state_machine import RerunStateMachine
 from megatron.core.transformer import TransformerConfig
-from megatron.core.transformer.pipeline_parallel_layer_layout import PipelineParallelLayerLayout
 from megatron.core.transformer.cuda_graph_config import (
     ALLOWED_INFERENCE_SCOPES,
     get_deprecated_cuda_graph_modules_migration,
@@ -23,23 +23,24 @@
     validate_deprecated_cuda_graph_modules_migration_inputs,
 )
 from megatron.core.transformer.enums import AttnBackend, CudaGraphModule, InferenceCudaGraphScope
+from megatron.core.transformer.pipeline_parallel_layer_layout import PipelineParallelLayerLayout
 from megatron.core.utils import (
     get_torch_version,
     is_flashinfer_min_version,
     is_te_min_version,
     is_torch_min_version,
 )
+from megatron.training.argument_utils import (  # noqa: F401 # pylint: disable=unused-import
+    ArgumentGroupFactory,
+    core_transformer_config_from_args,
+)
 from megatron.training.global_vars import set_global_variables
 from megatron.training.utils import (
     get_device_arch_version,
-    update_use_dist_ckpt,
     print_rank_0,
+    update_use_dist_ckpt,
     warn_rank_0,
 )
-from megatron.core.msc_utils import MultiStorageClientFeature
-
-from megatron.training.argument_utils import ArgumentGroupFactory, core_transformer_config_from_args  # noqa: F401 # pylint: disable=unused-import
-
 
 
 def add_megatron_arguments(parser: argparse.ArgumentParser):
@@ -398,8 +399,9 @@ def validate_args(args, defaults={}):
         'Currently only global and local checkpoints are supported'
     if args.non_persistent_ckpt_type == 'local':
         try:
-            from nvidia_resiliency_ext.checkpointing.local.ckpt_managers.local_manager import \
-                LocalCheckpointManager
+            from nvidia_resiliency_ext.checkpointing.local.ckpt_managers.local_manager import (
+                LocalCheckpointManager,
+            )
         except ModuleNotFoundError as e:
             raise RuntimeError('nvidia_resiliency_ext is required for local checkpointing') from e
 
@@ -740,8 +742,10 @@ def validate_args(args, defaults={}):
         )
 
     from megatron.core.models.hybrid.hybrid_layer_allocation import (
-        Symbols, parse_hybrid_pattern, get_hybrid_total_layer_count,
+        Symbols,
+        get_hybrid_total_layer_count,
         get_hybrid_total_pipeline_segment_count,
+        parse_hybrid_pattern,
     )
     sep = Symbols.MTP_SEPARATOR
 
@@ -948,20 +952,10 @@ def validate_args(args, defaults={}):
         if args.hybrid_layer_pattern is None:
             args.virtual_pipeline_model_parallel_size = None
 
-        if args.decoder_first_pipeline_num_layers is None and args.decoder_last_pipeline_num_layers is None:
-            # Divisibility check not applicable for T5 models which specify encoder_num_layers
-            # and decoder_num_layers, or for hybrid models using --hybrid-layer-pattern.
-            if args.num_layers is not None and args.hybrid_layer_pattern is None:
-                num_layers = args.num_layers
-
-                if args.account_for_embedding_in_pipeline_split:
-                    num_layers += 1
-
-                if args.account_for_loss_in_pipeline_split:
-                    num_layers += 1
-
-                assert num_layers % args.transformer_pipeline_model_parallel_size == 0, \
-                    'Number of layers should be divisible by the pipeline-model-parallel size'
+        # Defer layer-count divisibility to the selected model builder. Direct
+        # HybridModel architecture specs are Python objects constructed after
+        # CLI validation and may legally define uneven or empty explicit chunks.
+        # Conventional transformer builders retain their own divisibility checks.
 
     if args.virtual_pipeline_model_parallel_size is not None:
         if args.overlap_p2p_comm:
@@ -975,6 +969,11 @@ def validate_args(args, defaults={}):
                 'p2p sends and recvs between same 2 ranks per communication batch'
     else:
         # Overlap P2P communication is disabled if not using the interleaved schedule.
+        # Preserve the requested values because a Python HybridModel architecture
+        # can infer VPP only after this CLI-only validation pass.
+        if args.hybrid_layer_pattern is None:
+            args._overlap_p2p_comm_before_direct_vpp = args.overlap_p2p_comm
+            args._align_param_gather_before_direct_vpp = args.align_param_gather
         args.overlap_p2p_comm = False
         args.align_param_gather = False
         # Only print warning if PP size > 1.
@@ -1034,8 +1033,13 @@ def validate_args(args, defaults={}):
             '--overlap-param-gather-with-optimizer-step only supported with distributed optimizer'
         assert args.overlap_param_gather, \
             'Must use --overlap-param-gather-with-optimizer-step with --overlap-param-gather'
-        assert args.virtual_pipeline_model_parallel_size is not None, \
-            '--overlap-param-gather-with-optimizer-step only supported with interleaved pipeline parallelism'
+        if args.hybrid_layer_pattern is not None:
+            assert args.virtual_pipeline_model_parallel_size is not None, (
+                '--overlap-param-gather-with-optimizer-step only supported with '
+                'interleaved pipeline parallelism'
+            )
+        # Otherwise, the interleaved-pipeline requirement is checked in pretrain after
+        # Python model builders have had a chance to infer direct VPP splits.
         assert not args.use_dist_ckpt, \
             '--overlap-param-gather-with-optimizer-step not supported with distributed checkpointing yet'
 
@@ -2714,8 +2718,7 @@ def _add_rl_args(parser):
     return parser
 
 def _add_training_args(parser):
-    from megatron.training.config import TrainingConfig
-    from megatron.training.config import ProfilingConfig
+    from megatron.training.config import ProfilingConfig, TrainingConfig
 
     prof_factory = ArgumentGroupFactory(ProfilingConfig)
     prof_group = prof_factory.build_group(parser, "profiling")
@@ -3582,7 +3585,7 @@ def _add_kitchen_quantization_arguments(parser: argparse.ArgumentParser):
     If kitchen isn't available, nothing to do here, return unchanged parser
     """
     try:
-        from megatron.core.extensions.kitchen import KitchenSpecProvider, HAVE_KITCHEN
+        from megatron.core.extensions.kitchen import HAVE_KITCHEN, KitchenSpecProvider
 
     except (ImportError, ModuleNotFoundError):
         HAVE_KITCHEN = False
diff --git a/megatron/training/checkpointing.py b/megatron/training/checkpointing.py
index ee7e0077c14..369e3659279 100644
--- a/megatron/training/checkpointing.py
+++ b/megatron/training/checkpointing.py
@@ -3,6 +3,7 @@
 """Input/output checkpointing."""
 
 import contextlib
+import copy
 import inspect
 import multiprocessing
 import os
@@ -39,7 +40,10 @@
 from megatron.core.msc_utils import maybe_msc
 from megatron.core.num_microbatches_calculator import update_num_microbatches
 from megatron.core.optimizer import DistributedOptimizer
-from megatron.core.post_training.modelopt.checkpointing import save_modelopt_state, save_sharded_modelopt_state
+from megatron.core.post_training.modelopt.checkpointing import (
+    save_modelopt_state,
+    save_sharded_modelopt_state,
+)
 from megatron.core.rerun_state_machine import get_rerun_state_machine
 from megatron.core.utils import get_pg_rank, get_pg_size, unwrap_model
 from megatron.post_training.utils import print_distributed_quant_summary
@@ -1348,7 +1352,16 @@ def generate_state_dict(
 
     # Arguments, iteration, and model.
     state_dict = {}
-    state_dict['args'] = args
+    resolved_architecture = getattr(args, 'resolved_hybrid_architecture', None)
+    if getattr(resolved_architecture, "source", None) == "direct":
+        checkpoint_args = copy.copy(args)
+        # This runtime-only summary contains live ModuleSpec/config objects. Direct
+        # architecture recipes must be supplied by Python again on resume.
+        del checkpoint_args.resolved_hybrid_architecture
+        state_dict['args'] = checkpoint_args
+    else:
+        # Preserve the historical args object identity for legacy and non-hybrid checkpoints.
+        state_dict['args'] = args
     state_dict['checkpoint_version'] = 3.0
     if iteration is not None:
         state_dict['iteration'] = iteration
diff --git a/megatron/training/hybrid_metrics.py b/megatron/training/hybrid_metrics.py
new file mode 100644
index 00000000000..f4f05f29716
--- /dev/null
+++ b/megatron/training/hybrid_metrics.py
@@ -0,0 +1,75 @@
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
+"""Helpers for reporting metrics from a resolved hybrid architecture."""
+
+from dataclasses import dataclass
+from typing import Any, Iterator
+
+_LAYER_TYPE_ALIASES = {
+    "M": "mamba",
+    "G": "gdn",
+    "*": "attention",
+    "D": "dsa",
+    "+": "mla",
+    "-": "mlp",
+    "E": "moe",
+}
+
+
+def get_resolved_hybrid_architecture(args: Any) -> Any | None:
+    """Return a model-provided direct architecture, if one was installed."""
+
+    architecture = getattr(args, "resolved_hybrid_architecture", None)
+    if architecture is None or getattr(architecture, "source", None) != "direct":
+        return None
+    return architecture
+
+
+def get_hybrid_layer_type(layer: Any) -> str:
+    """Return the stable semantic type name for a resolved hybrid layer."""
+
+    layer_type = layer.layer_type
+    if not isinstance(layer_type, str):
+        layer_type = getattr(layer_type, "value", layer_type)
+    layer_type = str(layer_type)
+    return _LAYER_TYPE_ALIASES.get(layer_type, layer_type.lower())
+
+
+def iter_resolved_hybrid_layers(architecture: Any) -> Iterator[Any]:
+    """Iterate main layers followed by each repeated MTP-depth template."""
+
+    yield from architecture.main_layers
+    for _ in range(architecture.mtp_num_layers):
+        yield from architecture.mtp_layers
+
+
+@dataclass(frozen=True)
+class HybridMoEMetricMetadata:
+    """Arguments needed to size and normalize global MoE metric tensors."""
+
+    num_layers: int
+    moe_layer_freq: list[int]
+    mtp_num_layers: int
+    num_moe_layers: int
+
+
+def get_hybrid_moe_metric_metadata(architecture: Any) -> HybridMoEMetricMetadata:
+    """Derive MoE logging metadata from per-occurrence semantic layer types.
+
+    Direct hybrid models assign metric slots over the fully expanded global
+    architecture, including every layer in every MTP depth. Consequently MTP
+    has already been incorporated into ``num_layers`` and ``moe_layer_freq``;
+    ``mtp_num_layers`` is zero to disable the tracker's legacy implicit MTP
+    expansion.
+    """
+
+    layer_types = [
+        get_hybrid_layer_type(layer) for layer in iter_resolved_hybrid_layers(architecture)
+    ]
+    moe_layer_freq = [int(layer_type == "moe") for layer_type in layer_types]
+    return HybridMoEMetricMetadata(
+        num_layers=len(layer_types),
+        moe_layer_freq=moe_layer_freq,
+        mtp_num_layers=0,
+        num_moe_layers=sum(moe_layer_freq),
+    )
diff --git a/megatron/training/models/hybrid.py b/megatron/training/models/hybrid.py
index 287ca8ec2a3..f5366a85b0c 100644
--- a/megatron/training/models/hybrid.py
+++ b/megatron/training/models/hybrid.py
@@ -1,27 +1,32 @@
 # Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
 
 import logging
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Callable, ClassVar, Literal, override
 
 from megatron.core.distributed.distributed_data_parallel_config import DistributedDataParallelConfig
 from megatron.core.enums import ModelType
+from megatron.core.models.hybrid.hybrid_architecture import (
+    HybridLayerPattern,
+    resolve_hybrid_architecture,
+)
+from megatron.core.models.hybrid.hybrid_layer_specs import hybrid_inference_stack_spec
 from megatron.core.models.hybrid.hybrid_layer_specs import (
     hybrid_stack_spec as default_hybrid_stack_spec,
-    hybrid_inference_stack_spec,
 )
 from megatron.core.models.hybrid.hybrid_model import HybridModel
-from megatron.core.pipeline_parallel.utils import is_pp_first_stage, is_pp_last_stage
+from megatron.core.pipeline_parallel.utils import (
+    is_pp_first_stage,
+    is_pp_last_stage,
+    is_vp_first_stage,
+    is_vp_last_stage,
+)
 from megatron.core.post_training.modelopt.hybrid.model_specs import get_hybrid_stack_modelopt_spec
 from megatron.core.process_groups_config import ProcessGroupCollection
 from megatron.core.transformer.module import Float16Module, MegatronModule
 from megatron.core.transformer.spec_utils import ModuleSpec
 from megatron.core.transformer.transformer_config import TransformerConfig
-from megatron.training.models.base import (
-    ModelBuilder,
-    ModelConfig,
-    compose_hooks,
-)
+from megatron.training.models.base import ModelBuilder, ModelConfig, compose_hooks
 from megatron.training.models.dist_utils import unimodal_build_distributed_models
 from megatron.training.vocab_utils import calculate_padded_vocab_size
 
@@ -39,7 +44,8 @@ class HybridModelConfig(ModelConfig):
     on the embedded ``transformer`` config are accessible directly on this object
     via ``__getattr__``/``__setattr__`` proxying.
 
-    Supports hybrid architectures via ``hybrid_layer_pattern``
+    Direct ``layer_specs`` are the preferred architecture API. The legacy
+    ``hybrid_layer_pattern`` string remains supported for compatibility.
 
     Note:
         ``vocab_size`` must be set before passing this config to ``HybridModelBuilder``.
@@ -56,6 +62,8 @@ class HybridModelConfig(ModelConfig):
     hybrid_mlp_ratio: float = 0.0
     hybrid_override_pattern: str | None = None
     hybrid_layer_pattern: str | None = None
+    layer_specs: HybridLayerPattern | None = field(default=None, repr=False)
+    mtp_layer_specs: HybridLayerPattern | None = field(default=None, repr=False)
     seq_length: int = 8192
     # HybridModel with no attention has no need for position embeddings, so none is default
     position_embedding_type: Literal["learned_absolute", "rope", "none"] = "none"
@@ -109,6 +117,18 @@ def finalize(self) -> None:
         if hasattr(self.transformer, "finalize") and callable(self.transformer.finalize):
             self.transformer.finalize()
 
+    @override
+    def as_dict(self) -> dict[str, Any]:
+        """Serialize scalar configuration without live Python architecture objects.
+
+        The Python recipe must supply direct ``ModuleSpec`` trees again when resuming.
+        """
+
+        result = super().as_dict()
+        result.pop("layer_specs", None)
+        result.pop("mtp_layer_specs", None)
+        return result
+
 
 class HybridModelBuilder(ModelBuilder[HybridModel, HybridModelConfig]):
     """Builder to construct Megatron Core Hybrid models.
@@ -126,6 +146,101 @@ class HybridModelBuilder(ModelBuilder[HybridModel, HybridModelConfig]):
 
     def __init__(self, model_config: HybridModelConfig):
         super().__init__(model_config)
+        has_direct_specs = (
+            model_config.layer_specs is not None or model_config.mtp_layer_specs is not None
+        )
+        # Only direct descriptors need topology resolution before distributed construction.
+        # Legacy patterns retain the historical builder and runtime selector path.
+        if has_direct_specs:
+            self._hybrid_stack_spec = self._get_hybrid_stack_spec()
+            self._resolved_architecture = resolve_hybrid_architecture(
+                config=model_config.transformer,
+                hybrid_stack_spec=self._hybrid_stack_spec,
+                layer_specs=model_config.layer_specs,
+                mtp_layer_specs=model_config.mtp_layer_specs,
+                hybrid_layer_pattern=model_config.hybrid_layer_pattern,
+            )
+
+    @classmethod
+    def prepare_config_for_distributed_init(
+        cls, model_config: HybridModelConfig, args: Any
+    ) -> bool:
+        """Resolve split topology before Megatron initializes pipeline runtime state.
+
+        Direct Python specs are not present while command-line arguments are
+        validated, so their inferred VPP size must be copied to the runtime
+        namespace before ``initialize_model_parallel`` runs. Returns whether
+        direct architecture state was prepared.
+        """
+
+        if model_config.layer_specs is None and model_config.mtp_layer_specs is None:
+            return False
+
+        builder = cls(model_config)
+        resolved_architecture = getattr(builder, "_resolved_architecture", None)
+        if resolved_architecture is None:
+            return False
+
+        transformer = model_config.transformer
+        pp_size = transformer.pipeline_model_parallel_size
+        runtime_pp_size = getattr(args, "pipeline_model_parallel_size", pp_size)
+        if runtime_pp_size != pp_size:
+            raise ValueError(
+                "HybridModelConfig.transformer.pipeline_model_parallel_size must match the "
+                f"runtime pipeline topology; got {pp_size} != {runtime_pp_size}."
+            )
+
+        inferred_vp_size = transformer.virtual_pipeline_model_parallel_size
+        runtime_vp_size = getattr(args, "virtual_pipeline_model_parallel_size", None)
+        if runtime_vp_size is not None and runtime_vp_size != inferred_vp_size:
+            raise ValueError(
+                "Hybrid architecture splits disagree with the runtime virtual pipeline "
+                f"topology; got {inferred_vp_size} != {runtime_vp_size}."
+            )
+        args.virtual_pipeline_model_parallel_size = inferred_vp_size
+
+        # Argument validation temporarily disables interleaved-only options when
+        # direct Python split nodes are not available yet. Restore the user's
+        # requested settings now that those nodes have inferred VPP.
+        if inferred_vp_size is not None and runtime_vp_size is None:
+            requested_overlap = getattr(
+                args,
+                "_overlap_p2p_comm_before_direct_vpp",
+                getattr(args, "overlap_p2p_comm", False),
+            )
+            requested_align = getattr(
+                args,
+                "_align_param_gather_before_direct_vpp",
+                getattr(args, "align_param_gather", False),
+            )
+            if pp_size == 2 and not requested_overlap:
+                raise ValueError(
+                    "Direct PP2/VPP interleaving requires P2P communication overlap; "
+                    "remove --no-overlap-p2p-communication."
+                )
+            args.overlap_p2p_comm = requested_overlap
+            args.align_param_gather = requested_align
+            if hasattr(args, "batch_p2p_comm"):
+                args.batch_p2p_comm = not requested_overlap
+            transformer.overlap_p2p_comm = requested_overlap
+            transformer.batch_p2p_comm = not requested_overlap
+
+        return True
+
+    def _get_hybrid_stack_spec(self) -> ModuleSpec:
+        """Select the stack implementation used by every local model chunk."""
+
+        hybrid_stack_spec = self._model_config.hybrid_stack_spec
+        if hybrid_stack_spec is not None:
+            return hybrid_stack_spec
+        if self._model_config.transformer.transformer_impl == "inference_optimized":
+            return hybrid_inference_stack_spec
+        if self._model_config.restore_modelopt_state:
+            return get_hybrid_stack_modelopt_spec(
+                local_core_attention=False,
+                remap_te_layernorm=False,
+            )
+        return default_hybrid_stack_spec
 
     def build_model(
         self,
@@ -145,20 +260,25 @@ def build_model(
         Returns:
             The constructed model
 
-        Note:
-            Virtual pipeline model parallelism is not supported for Hybrid models.
         """
-        hybrid_stack_spec = self._model_config.hybrid_stack_spec
-        if hybrid_stack_spec is None:
-            if self._model_config.transformer.transformer_impl == "inference_optimized":
-                hybrid_stack_spec = hybrid_inference_stack_spec
-            elif self._model_config.restore_modelopt_state:
-                hybrid_stack_spec = get_hybrid_stack_modelopt_spec(
-                    local_core_attention=False,
-                    remap_te_layernorm=False,
-                )
-            else:
-                hybrid_stack_spec = default_hybrid_stack_spec
+        # Re-resolve if a caller selected a different implementation spec after
+        # builder construction (for example, modelopt or optimized inference).
+        hybrid_stack_spec = self._get_hybrid_stack_spec()
+        has_direct_specs = (
+            self._model_config.layer_specs is not None
+            or self._model_config.mtp_layer_specs is not None
+        )
+        if has_direct_specs and hybrid_stack_spec is not getattr(
+            self, "_hybrid_stack_spec", None
+        ):
+            self._hybrid_stack_spec = hybrid_stack_spec
+            self._resolved_architecture = resolve_hybrid_architecture(
+                config=self._model_config.transformer,
+                hybrid_stack_spec=hybrid_stack_spec,
+                layer_specs=self._model_config.layer_specs,
+                mtp_layer_specs=self._model_config.mtp_layer_specs,
+                hybrid_layer_pattern=self._model_config.hybrid_layer_pattern,
+            )
 
         assert self._model_config.vocab_size is not None, "vocab_size must be configured before calling build_model()"
         if self._model_config.should_pad_vocab:
@@ -170,8 +290,35 @@ def build_model(
         else:
             padded_vocab_size = self._model_config.vocab_size
 
-        pre_process = pre_process if pre_process is not None else is_pp_first_stage(pg_collection.pp)
-        post_process = post_process if post_process is not None else is_pp_last_stage(pg_collection.pp)
+        resolved_architecture = getattr(self, "_resolved_architecture", None)
+        is_direct = resolved_architecture is not None
+        if is_direct:
+            vp_size = self._model_config.transformer.virtual_pipeline_model_parallel_size
+            if vp_size is not None and vp_stage is None:
+                # A single-chunk build defaults to the first virtual stage, matching
+                # ResolvedHybridArchitecture.select_segment's public semantics.
+                vp_stage = 0
+            pre_process = (
+                pre_process
+                if pre_process is not None
+                else is_pp_first_stage(pg_collection.pp) and is_vp_first_stage(vp_stage, vp_size)
+            )
+            post_process = (
+                post_process
+                if post_process is not None
+                else is_pp_last_stage(pg_collection.pp) and is_vp_last_stage(vp_stage, vp_size)
+            )
+        else:
+            pre_process = (
+                pre_process if pre_process is not None else is_pp_first_stage(pg_collection.pp)
+            )
+            post_process = (
+                post_process if post_process is not None else is_pp_last_stage(pg_collection.pp)
+            )
+
+        direct_architecture_kwargs = (
+            {"resolved_hybrid_architecture": resolved_architecture} if is_direct else {}
+        )
         return HybridModel(
             config=self._model_config.transformer,
             hybrid_stack_spec=hybrid_stack_spec,
@@ -189,6 +336,7 @@ def build_model(
             post_process=post_process,
             pg_collection=pg_collection,
             vp_stage=vp_stage,
+            **direct_architecture_kwargs,
         )
 
     def build_distributed_models(
diff --git a/megatron/training/training.py b/megatron/training/training.py
index 7d5b954dc61..a200aeac9f3 100644
--- a/megatron/training/training.py
+++ b/megatron/training/training.py
@@ -146,6 +146,12 @@
 from megatron.training.config import FaultInjectorConfig
 from megatron.training.config.container import PretrainConfigContainer
 from megatron.training.datasets.data_samplers import build_pretraining_data_loader
+from megatron.training.hybrid_metrics import (
+    get_hybrid_layer_type,
+    get_hybrid_moe_metric_metadata,
+    get_resolved_hybrid_architecture,
+    iter_resolved_hybrid_layers,
+)
 from megatron.training.initialize import (
     initialize_megatron,
     set_jit_fusion_options,
@@ -471,16 +477,21 @@ def attn_layer_flops(
             + 2 * seqlen_squared_sum * hidden_size * p
         )
 
-    def mamba_layer_flops(total_tokens, hidden_size, state_dim=16,
-                          head_dim=64, num_groups=1, num_heads=128):
+    def mamba_layer_flops(
+        total_tokens,
+        hidden_size,
+        state_dim=16,
+        head_dim=64,
+        num_groups=1,
+        num_heads=128,
+        d_in=None,
+    ):
         """Calculate FLOPs for a Mamba layer."""
         # Note (rwaleffe): flops estimate for scan should be updated based on new SSD kernels,
         # but small percent of overall layer flops
-        d_in = 2 * hidden_size
-        if num_heads:
-            nheads = num_heads
-        else:
-            nheads = d_in // head_dim
+        if d_in is None:
+            d_in = 2 * hidden_size
+        nheads = num_heads if num_heads else d_in // head_dim
         return (
             (
                 2
@@ -547,6 +558,127 @@ def hybrid_flops(total_tokens, seqlen_squared_sum, hidden_size,
         )
         return flops_fwd * 3
 
+    def resolved_hybrid_flops(architecture):
+        """Calculate hybrid FLOPs from every resolved layer occurrence."""
+
+        flops_fwd = 0
+        mtp_num_layers = architecture.mtp_num_layers
+        for layer in iter_resolved_hybrid_layers(architecture):
+            config = layer.config
+            layer_type = get_hybrid_layer_type(layer)
+            hidden_size = config.hidden_size
+
+            if layer_type == "attention":
+                num_heads = config.num_attention_heads
+                num_query_groups = config.num_query_groups or num_heads
+                flops_fwd += attn_layer_flops(
+                    total_real_tokens_in_batch,
+                    seqlen_squared_sum_in_batch,
+                    hidden_size,
+                    num_heads,
+                    gqa=num_query_groups != num_heads,
+                    gqa_groups=num_query_groups,
+                    kv_channels=config.kv_channels,
+                )
+            elif layer_type in {"mla", "dsa"}:
+                if config.q_lora_rank is None:
+                    q_term = (
+                        config.hidden_size
+                        * config.num_attention_heads
+                        * (config.qk_head_dim + config.qk_pos_emb_head_dim)
+                    )
+                else:
+                    q_term = config.q_lora_rank * (
+                        config.hidden_size
+                        + config.num_attention_heads
+                        * (config.qk_head_dim + config.qk_pos_emb_head_dim)
+                        + 1
+                    )
+                projection_term = (
+                    q_term
+                    + config.kv_lora_rank
+                    * (
+                        config.hidden_size
+                        + config.num_attention_heads
+                        * (config.qk_head_dim + config.v_head_dim)
+                        + 1
+                    )
+                    + config.hidden_size * config.qk_pos_emb_head_dim
+                    + config.num_attention_heads * config.v_head_dim * config.hidden_size
+                )
+                core_term = config.num_attention_heads * (
+                    config.qk_head_dim + config.qk_pos_emb_head_dim + config.v_head_dim
+                )
+                flops_fwd += (
+                    2 * total_real_tokens_in_batch * projection_term
+                    + seqlen_squared_sum_in_batch * core_term
+                )
+            elif layer_type == "mamba":
+                flops_fwd += mamba_layer_flops(
+                    total_real_tokens_in_batch,
+                    hidden_size,
+                    state_dim=config.mamba_state_dim,
+                    head_dim=config.mamba_head_dim,
+                    num_groups=config.mamba_num_groups,
+                    num_heads=config.mamba_num_heads,
+                    d_in=(
+                        config.mamba_num_heads * config.mamba_head_dim
+                        if config.mamba_num_heads
+                        else 2 * hidden_size
+                    ),
+                )
+            elif layer_type == "mlp":
+                flops_fwd += mlp_layer_flops(
+                    total_real_tokens_in_batch,
+                    hidden_size,
+                    expansion=config.ffn_hidden_size / hidden_size,
+                    swiglu=config.gated_linear_unit,
+                )
+            elif layer_type == "moe":
+                flops_fwd += moe_layer_flops(
+                    total_real_tokens_in_batch,
+                    hidden_size,
+                    moe_ffn_hidden_size=config.moe_ffn_hidden_size or config.ffn_hidden_size,
+                    shared_expert_ffn_hidden_size=(config.moe_shared_expert_intermediate_size or 0),
+                    num_experts_routed_to=config.moe_router_topk,
+                    moe_latent_size=config.moe_latent_size,
+                    swiglu=config.gated_linear_unit,
+                )
+            elif layer_type == "gdn":
+                flops_fwd += gdn_layer_flops(
+                    total_real_tokens_in_batch,
+                    hidden_size,
+                    qk_head_dim=config.linear_key_head_dim or 128,
+                    v_head_dim=config.linear_value_head_dim or 128,
+                    num_qk_heads=config.linear_num_key_heads or 16,
+                    num_v_heads=config.linear_num_value_heads or 32,
+                    conv_kernel_dim=config.linear_conv_kernel_dim or 4,
+                )
+            else:
+                raise ValueError(
+                    f"FLOPs calculation is not implemented for resolved hybrid "
+                    f"layer type {layer_type!r}."
+                )
+
+        # Each MTP depth adds two norms, a final norm, and the eh projection.
+        # Layer work above already includes every repeated MTP template layer.
+        flops_fwd += (
+            2
+            * total_real_tokens_in_batch
+            * mtp_num_layers
+            * (3 * args.hidden_size + 2 * args.hidden_size * args.hidden_size)
+        )
+
+        # The main decoder and every MTP depth use an output head.
+        flops_fwd += (
+            2
+            * total_real_tokens_in_batch
+            * args.hidden_size
+            * args.padded_vocab_size
+            * (1 + mtp_num_layers)
+        )
+        return flops_fwd * 3
+
     def transformer_flops():
         """Calculate FLOPs for a standard Transformer model."""
         # TODO(helenn/dnarayanan): Refactor this to reuse the helper methods.
@@ -848,6 +980,10 @@ def transformer_flops():
 
     # Main entrypoint for FLOPs calculation.
     if is_hybrid_model(args):
+        resolved_architecture = get_resolved_hybrid_architecture(args)
+        if resolved_architecture is not None:
+            return resolved_hybrid_flops(resolved_architecture)
+
         # Calculate the number of each type of layer.
         from operator import itemgetter
 
@@ -1085,6 +1221,45 @@ def pretrain(
     ft_integration.setup()
     timestamp_after_in_job_setup = time.time()
 
+    # Model-defined split nodes may infer VPP after command-line validation.
+    # Give builders a chance to synchronize that topology before MPU state is
+    # initialized and before schedules/data iterators consult the runtime args.
+    runtime_args = get_args()
+    model_config = getattr(cfg_container, "model", None)
+    prepared_direct_architecture = False
+    if model_config is not None:
+        builder_cls = model_config.get_builder_cls()
+        prepare_for_distributed_init = getattr(
+            builder_cls, "prepare_config_for_distributed_init", None
+        )
+        if prepare_for_distributed_init is not None:
+            prepared_direct_architecture = bool(
+                prepare_for_distributed_init(model_config, runtime_args)
+            )
+
+    # Synchronize config objects that were materialized after CLI validation but
+    # before a direct architecture could infer VPP.
+    if (
+        prepared_direct_architecture
+        and getattr(runtime_args, "virtual_pipeline_model_parallel_size", None) is not None
+        and getattr(cfg_container, "ddp", None) is not None
+    ):
+        cfg_container.ddp.align_param_gather = runtime_args.align_param_gather
+    if (
+        getattr(runtime_args, "overlap_param_gather_with_optimizer_step", False)
+        and getattr(runtime_args, "virtual_pipeline_model_parallel_size", None) is None
+    ):
+        raise AssertionError(
+            "--overlap-param-gather-with-optimizer-step only supported with "
+            "interleaved pipeline parallelism"
+        )
+    for deferred_attr in (
+        "_overlap_p2p_comm_before_direct_vpp",
+        "_align_param_gather_before_direct_vpp",
+    ):
+        if hasattr(runtime_args, deferred_attr):
+            delattr(runtime_args, deferred_attr)
+
     # Multimodal MiMo seeds each module's RNG in its builder; a plain collection seeds stock here.
     skip_random_seed = isinstance(pg_collection, MultiModuleProcessGroupCollection)
 
@@ -2110,6 +2285,23 @@ def _build_model_wrapper(wrap_with_ddp: bool):
     model = _build_model_wrapper(wrap_with_ddp)
     unwrapped_model = unwrap_model(model)
 
+    # Resolved hybrid architectures are model-construction artifacts rather
+    # than command-line arguments. Make the shared global summary available to
+    # training metrics after all physical/virtual chunks have been built.
+    if hasattr(args, "resolved_hybrid_architecture"):
+        delattr(args, "resolved_hybrid_architecture")
+    model_chunks = unwrapped_model if isinstance(unwrapped_model, list) else [unwrapped_model]
+    for model_chunk in model_chunks:
+        resolved_hybrid_architecture = getattr(
+            model_chunk, "resolved_hybrid_architecture", None
+        )
+        if (
+            resolved_hybrid_architecture is not None
+            and resolved_hybrid_architecture.source == "direct"
+        ):
+            args.resolved_hybrid_architecture = resolved_hybrid_architecture
+            break
+
     # Classify each GTP param's prefetch chain after model build + DDP wrap, before the
     # first forward. Placed here (not in get_model) so it also covers the config-container
     # builder path.
@@ -2792,7 +2984,18 @@ def training_log(
                 wandb_writer.log({'max_attention_logit': max_attention_logit}, iteration)
     # Log MoE metrics.
     moe_log_string = ""
-    if args.num_experts is not None:
+    resolved_architecture = get_resolved_hybrid_architecture(args)
+    resolved_moe_metadata = (
+        get_hybrid_moe_metric_metadata(resolved_architecture)
+        if resolved_architecture is not None
+        else None
+    )
+    has_moe_layers = (
+        resolved_moe_metadata.num_moe_layers > 0
+        if resolved_moe_metadata is not None
+        else args.num_experts is not None
+    )
+    if has_moe_layers:
         moe_loss_scale = 1 / get_num_microbatches()
         track_names = []
         if "aux_loss" in args.moe_router_load_balancing_type:
@@ -2804,7 +3007,13 @@ def training_log(
         if args.moe_z_loss_coeff is not None:
             track_names.append("z_loss")
 
-        if is_hybrid_model(args):
+        if resolved_architecture is not None:
+            assert resolved_moe_metadata is not None
+            moe_metadata = resolved_moe_metadata
+            layers = moe_metadata.num_layers
+            moe_layer_freq = moe_metadata.moe_layer_freq
+            mtp_num_layers = moe_metadata.mtp_num_layers
+        elif is_hybrid_model(args):
             from operator import itemgetter
 
             from megatron.core.ssm.mamba_hybrid_layer_allocation import (
@@ -2812,8 +3021,12 @@ def training_log(
                 get_hybrid_layer_counts,
             )
             layers = itemgetter(Symbols.MOE)(get_hybrid_layer_counts(args.hybrid_layer_pattern))
+            moe_layer_freq = args.moe_layer_freq
+            mtp_num_layers = args.mtp_num_layers
         else:
             layers = args.num_layers
+            moe_layer_freq = args.moe_layer_freq
+            mtp_num_layers = args.mtp_num_layers
 
         moe_log_string = get_moe_metrics_tracker().report(
             loss_scale=moe_loss_scale,
@@ -2824,14 +3037,19 @@ def training_log(
             force_initialize=True,
             track_names=track_names,
             num_layers=layers,
-            moe_layer_freq=args.moe_layer_freq,
-            mtp_num_layers=args.mtp_num_layers,
+            moe_layer_freq=moe_layer_freq,
+            mtp_num_layers=mtp_num_layers,
             pg_collection=pg_collection,
             total_loss_dict=total_loss_dict,
         )
 
     # Log MTP metrics.
-    if args.mtp_num_layers is not None:
+    has_mtp_metrics = (
+        resolved_architecture.mtp_num_layers > 0
+        if resolved_architecture is not None
+        else args.mtp_num_layers is not None
+    )
+    if has_mtp_metrics:
         mtp_loss_scale = 1 / get_num_microbatches()
         MTPLossLoggingHelper.track_mtp_metrics(
             mtp_loss_scale, iteration, writer, wandb_writer, total_loss_dict
diff --git a/megatron/training/utils/common_utils.py b/megatron/training/utils/common_utils.py
index 30617ef9b4c..cee4bbfba78 100644
--- a/megatron/training/utils/common_utils.py
+++ b/megatron/training/utils/common_utils.py
@@ -12,10 +12,10 @@
 
 import torch
 
-from megatron.core.msc_utils import maybe_msc
 from megatron.core._rank_utils import safe_get_rank as _safe_get_rank
 from megatron.core._slurm_utils import resolve_slurm_local_rank
 from megatron.core.dist_checkpointing.strategies.nvrx import has_nvrx_async_support
+from megatron.core.msc_utils import maybe_msc
 
 try:
     from transformer_engine.pytorch.optimizers import multi_tensor_applier, multi_tensor_l2norm
@@ -48,7 +48,6 @@
 from megatron.training import get_adlr_autoresume, get_args, get_timers
 
 
-
 def _compute_norm_2(params_list):
     """Compute squared L2 norm of a list of tensors. Returns a CUDA scalar."""
     if len(params_list) > 0:
@@ -476,7 +475,11 @@ def print_rank_last(message):
 
 def is_hybrid_model(args):
     """Returns True if the model is a hybrid Mamba-Transformer model."""
-    return args.hybrid_layer_pattern is not None
+    resolved_architecture = getattr(args, 'resolved_hybrid_architecture', None)
+    return (
+        getattr(args, 'hybrid_layer_pattern', None) is not None
+        or getattr(resolved_architecture, 'source', None) == 'direct'
+    )
 
 
 def is_gtp_remat_active(args):
diff --git a/tests/unit_tests/test_num_floating_point_operations.py b/tests/unit_tests/test_num_floating_point_operations.py
index f358ada8fc1..e17e1751880 100644
--- a/tests/unit_tests/test_num_floating_point_operations.py
+++ b/tests/unit_tests/test_num_floating_point_operations.py
@@ -18,6 +18,7 @@
 import torch
 
 import megatron.training.training as training_module
+from megatron.training.hybrid_metrics import get_hybrid_moe_metric_metadata
 from megatron.training.training import (
     consume_seqlen_stats_in_iteration,
     num_floating_point_operations,
@@ -99,6 +100,58 @@ def _make_hybrid_args(*, num_layers=4, hidden_size=512, num_attention_heads=8, s
     return args
 
 
+def _make_resolved_layer(layer_type, **overrides):
+    config_values = {
+        "hidden_size": 64,
+        "num_attention_heads": 4,
+        "num_query_groups": 2,
+        "kv_channels": 8,
+        "mamba_state_dim": 16,
+        "mamba_head_dim": 8,
+        "mamba_num_groups": 2,
+        "mamba_num_heads": 8,
+        "ffn_hidden_size": 160,
+        "gated_linear_unit": True,
+        "moe_ffn_hidden_size": 96,
+        "moe_shared_expert_intermediate_size": 32,
+        "moe_router_topk": 2,
+        "moe_latent_size": None,
+        "linear_key_head_dim": 16,
+        "linear_value_head_dim": 16,
+        "linear_num_key_heads": 4,
+        "linear_num_value_heads": 4,
+        "linear_conv_kernel_dim": 4,
+    }
+    config_values.update(overrides)
+    return SimpleNamespace(layer_type=layer_type, config=SimpleNamespace(**config_values))
+
+
+def _make_resolved_hybrid_args():
+    args = _make_gpt_args(
+        num_layers=4, hidden_size=64, num_attention_heads=4, seq_length=16, padded_vocab_size=100
+    )
+    # Direct metrics take MTP depth from the resolved architecture, not from
+    # legacy command-line state.
+    args.mtp_num_layers = None
+    args.resolved_hybrid_architecture = SimpleNamespace(
+        source="direct",
+        mtp_num_layers=2,
+        main_layers=(
+            _make_resolved_layer("attention"),
+            _make_resolved_layer("mamba"),
+            _make_resolved_layer("mlp"),
+            _make_resolved_layer("moe"),
+        ),
+        mtp_layers=(
+            _make_resolved_layer(
+                "attention", num_attention_heads=8, num_query_groups=1, kv_channels=4
+            ),
+            _make_resolved_layer("moe", moe_ffn_hidden_size=224, moe_router_topk=3),
+        ),
+    )
+    return args
+
+
 class TestBSHDBackwardCompat:
     """For unpacked BSHD, the new optional arg must not change the result."""
 
@@ -255,6 +308,131 @@ def test_hybrid_attention_layers_count(self):
         assert flops_doubled - flops_bshd == expected_delta
 
 
+class TestResolvedHybridMetrics:
+    """Direct hybrid metrics use every occurrence's resolved configuration."""
+
+    def test_heterogeneous_layer_flops(self):
+        args = _make_resolved_hybrid_args()
+        batch_size = 2
+        total_tokens = batch_size * args.seq_length
+        seqlen_squared_sum = batch_size * args.seq_length**2
+
+        def attention_flops(config):
+            p = config.kv_channels * config.num_attention_heads / config.hidden_size
+            return (
+                4
+                * total_tokens
+                * config.hidden_size
+                * p
+                * (
+                    config.hidden_size
+                    + config.hidden_size * (config.num_query_groups / config.num_attention_heads)
+                )
+                + 2 * seqlen_squared_sum * config.hidden_size * p
+            )
+
+        def mamba_flops(config):
+            # MambaMixer derives d_inner from explicit per-occurrence head geometry.
+            d_in = config.mamba_num_heads * config.mamba_head_dim
+            return (
+                2
+                * total_tokens
+                * config.hidden_size
+                * (
+                    2 * d_in
+                    + 2 * config.mamba_num_groups * config.mamba_state_dim
+                    + config.mamba_num_heads
+                )
+                + 7 * total_tokens * d_in * config.mamba_state_dim
+                + 2 * total_tokens * d_in * config.hidden_size
+            )
+
+        def mlp_flops(config):
+            return 4 * 1.5 * total_tokens * config.hidden_size * config.ffn_hidden_size
+
+        def moe_flops(config):
+            return (
+                4
+                * total_tokens
+                * config.hidden_size
+                * config.moe_ffn_hidden_size
+                * config.moe_router_topk
+                * 1.5
+                + 4
+                * total_tokens
+                * config.hidden_size
+                * config.moe_shared_expert_intermediate_size
+                * 1.5
+            )
+
+        architecture = args.resolved_hybrid_architecture
+        main = architecture.main_layers
+        mtp = architecture.mtp_layers
+        expected_forward = (
+            attention_flops(main[0].config)
+            + mamba_flops(main[1].config)
+            + mlp_flops(main[2].config)
+            + moe_flops(main[3].config)
+            + architecture.mtp_num_layers
+            * (attention_flops(mtp[0].config) + moe_flops(mtp[1].config))
+            + 2
+            * total_tokens
+            * architecture.mtp_num_layers
+            * (3 * args.hidden_size + 2 * args.hidden_size**2)
+            + 2
+            * total_tokens
+            * args.hidden_size
+            * args.padded_vocab_size
+            * (1 + architecture.mtp_num_layers)
+        )
+
+        actual = num_floating_point_operations(args, batch_size)
+
+        assert actual == pytest.approx(expected_forward * 3)
+
+    def test_moe_metadata_uses_expanded_metric_slots(self):
+        args = _make_resolved_hybrid_args()
+
+        metadata = get_hybrid_moe_metric_metadata(args.resolved_hybrid_architecture)
+
+        assert metadata.num_layers == 8
+        assert metadata.moe_layer_freq == [0, 0, 0, 1, 0, 1, 0, 1]
+        assert metadata.mtp_num_layers == 0
+        assert metadata.num_moe_layers == 3
+
+
+class TestLegacyHybridMetricsCompatibility:
+    """Legacy string patterns retain their historical args-based metric formulas."""
+
+    def test_legacy_mamba_flops_keep_model_global_d_in(self):
+        args = _make_hybrid_args(num_layers=1)
+        args.hybrid_layer_pattern = "M"
+        batch_size = 2
+        total_tokens = batch_size * args.seq_length
+        d_in = 2 * args.hidden_size
+        expected_forward = (
+            2
+            * total_tokens
+            * args.hidden_size
+            * (2 * d_in + 2 * args.mamba_num_groups * args.mamba_state_dim + args.mamba_num_heads)
+            + 7 * total_tokens * d_in * args.mamba_state_dim
+            + 2 * total_tokens * d_in * args.hidden_size
+            + 2 * total_tokens * args.hidden_size * args.padded_vocab_size
+        )
+
+        assert num_floating_point_operations(args, batch_size) == expected_forward * 3
+
+    def test_legacy_resolved_summary_does_not_select_direct_metric_path(self):
+        args = _make_hybrid_args()
+        batch_size = 2
+        expected = num_floating_point_operations(args, batch_size)
+        args.resolved_hybrid_architecture = SimpleNamespace(
+            source="legacy", main_layers=(), mtp_layers=(), mtp_num_layers=0
+        )
+
+        assert num_floating_point_operations(args, batch_size) == expected
+
+
 class TestPaddingRemoval:
     """``total_real_tokens_in_batch`` removes padding from token-linear FLOPs.
 
diff --git a/tests/unit_tests/training/models/test_hybrid_builder.py b/tests/unit_tests/training/models/test_hybrid_builder.py
index 9984e224ce3..771f48d8780 100644
--- a/tests/unit_tests/training/models/test_hybrid_builder.py
+++ b/tests/unit_tests/training/models/test_hybrid_builder.py
@@ -1,11 +1,13 @@
 # Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
 
+from types import SimpleNamespace
 from unittest.mock import Mock, call, patch
 
 import pytest
 
 from megatron.core.transformer import ModuleSpec
 from megatron.core.transformer.transformer_config import TransformerConfig
+from megatron.training.models.base import ModelConfig
 from megatron.training.models.hybrid import HybridModelBuilder, HybridModelConfig
 
 # ---------------------------------------------------------------------------
@@ -74,6 +76,14 @@ def test_hybrid_stack_spec_default_is_none(self):
         config = _make_hybrid_config()
         assert config.hybrid_stack_spec is None
 
+    def test_legacy_as_dict_matches_the_historical_field_set(self):
+        config = _make_hybrid_config(hybrid_layer_pattern="M*")
+        historical = ModelConfig.as_dict(config)
+        historical.pop("layer_specs")
+        historical.pop("mtp_layer_specs")
+
+        assert config.as_dict() == historical
+
 
 class TestHybridModelConfigGetAttr:
     """Tests for HybridModelConfig.__getattr__ — direct access vs. TransformerConfig proxy."""
@@ -176,6 +186,39 @@ def setup_method(self):
     def test_stores_model_config(self):
         assert self.builder._model_config is self.config
 
+    def test_legacy_pattern_is_not_pre_resolved_or_allowed_to_mutate_topology(self):
+        transformer = _make_transformer(pipeline_model_parallel_size=2)
+        config = _make_hybrid_config(transformer=transformer, hybrid_layer_pattern="M|M|M|M")
+
+        builder = HybridModelBuilder(config)
+
+        assert not hasattr(builder, "_resolved_architecture")
+        assert not hasattr(builder, "_hybrid_stack_spec")
+        assert transformer.virtual_pipeline_model_parallel_size is None
+
+    def test_legacy_builder_init_does_not_select_or_resolve_a_stack(self):
+        config = _make_hybrid_config(hybrid_layer_pattern="M*")
+
+        with (
+            patch.object(HybridModelBuilder, "_get_hybrid_stack_spec") as get_stack_spec,
+            patch("megatron.training.models.hybrid.resolve_hybrid_architecture") as resolve,
+        ):
+            builder = HybridModelBuilder(config)
+
+        assert not hasattr(builder, "_hybrid_stack_spec")
+        assert not hasattr(builder, "_resolved_architecture")
+        get_stack_spec.assert_not_called()
+        resolve.assert_not_called()
+
+    def test_legacy_config_is_ignored_by_direct_distributed_preparation(self):
+        config = _make_hybrid_config(hybrid_layer_pattern="M*")
+        args = SimpleNamespace(virtual_pipeline_model_parallel_size=None)
+
+        prepared = HybridModelBuilder.prepare_config_for_distributed_init(config, args)
+
+        assert prepared is False
+        assert args.virtual_pipeline_model_parallel_size is None
+
 
 class TestHybridModelBuilderBuildModel:
     """Tests for HybridModelBuilder.build_model() — spec resolution, vocab padding, pp-stage inference, and MCoreHybridModel kwargs."""
@@ -319,6 +362,24 @@ def test_infers_post_process_from_pg(self, mock_model, mock_first, mock_last, *_
         mock_last.assert_called_once_with(self.pg.pp)
         assert mock_model.call_args.kwargs["post_process"] is True
 
+    @patch("megatron.training.models.hybrid.is_vp_last_stage")
+    @patch("megatron.training.models.hybrid.is_vp_first_stage")
+    @patch("megatron.training.models.hybrid.is_pp_last_stage", return_value=True)
+    @patch("megatron.training.models.hybrid.is_pp_first_stage", return_value=False)
+    @patch("megatron.training.models.hybrid.HybridModel")
+    def test_legacy_builder_keeps_pp_only_defaults_when_vp_stage_is_supplied(
+        self, mock_model, _mock_first, _mock_last, mock_vp_first, mock_vp_last
+    ):
+        self.builder.build_model(self.pg, vp_stage=1)
+
+        kwargs = mock_model.call_args.kwargs
+        assert kwargs["vp_stage"] == 1
+        assert kwargs["pre_process"] is False
+        assert kwargs["post_process"] is True
+        assert "resolved_hybrid_architecture" not in kwargs
+        mock_vp_first.assert_not_called()
+        mock_vp_last.assert_not_called()
+
     @patch("megatron.training.models.hybrid.calculate_padded_vocab_size")
     @patch("megatron.training.models.hybrid.is_pp_last_stage", return_value=True)
     @patch("megatron.training.models.hybrid.is_pp_first_stage", return_value=True)
@@ -327,7 +388,7 @@ def test_config_params_passed_to_mcore(self, mock_model, *_):
         config = _make_hybrid_config(
             vocab_size=32000,
             seq_length=4096,
-            hybrid_layer_pattern="M-A-",
+            hybrid_layer_pattern="M*",
             fp16_lm_cross_entropy=True,
             parallel_output=False,
             share_embeddings_and_output_weights=True,
@@ -343,7 +404,7 @@ def test_config_params_passed_to_mcore(self, mock_model, *_):
         assert kw["config"] is config.transformer
         assert kw["vocab_size"] == 32000
         assert kw["max_sequence_length"] == 4096
-        assert kw["hybrid_layer_pattern"] == "M-A-"
+        assert kw["hybrid_layer_pattern"] == "M*"
         assert kw["fp16_lm_cross_entropy"] is True
         assert kw["parallel_output"] is False
         assert kw["share_embeddings_and_output_weights"] is True
@@ -353,6 +414,7 @@ def test_config_params_passed_to_mcore(self, mock_model, *_):
         assert kw["seq_len_interpolation_factor"] is None
         assert kw["pg_collection"] is pg
         assert kw["vp_stage"] is None
+        assert "resolved_hybrid_architecture" not in kw
 
 
 class TestHybridModelBuilderBuildDistributedModels:
diff --git a/tests/unit_tests/training/models/test_hybrid_direct_builder.py b/tests/unit_tests/training/models/test_hybrid_direct_builder.py
new file mode 100644
index 00000000000..a677a0fa56d
--- /dev/null
+++ b/tests/unit_tests/training/models/test_hybrid_direct_builder.py
@@ -0,0 +1,305 @@
+# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+
+"""Focused tests for direct HybridModelConfig and PP/VPP builder behavior."""
+
+import sys
+from copy import deepcopy
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pytest
+import torch
+
+from megatron.core.models.hybrid.hybrid_architecture import HybridLayerSpec, PipelineSplit
+from megatron.core.models.hybrid.hybrid_layer_specs import hybrid_stack_spec
+from megatron.core.transformer import TransformerConfig
+from megatron.training.argument_utils import core_transformer_config_from_args
+from megatron.training.arguments import parse_args, validate_args
+from megatron.training.checkpointing import generate_state_dict
+from megatron.training.models.hybrid import HybridModelBuilder, HybridModelConfig
+
+
+def _make_transformer() -> TransformerConfig:
+    return TransformerConfig(
+        num_layers=4,
+        hidden_size=128,
+        num_attention_heads=1,
+        pipeline_model_parallel_size=2,
+        pipeline_dtype=torch.float32,
+    )
+
+
+def _make_layer(config: TransformerConfig) -> HybridLayerSpec:
+    return HybridLayerSpec(
+        module_spec=hybrid_stack_spec.submodules.mamba_layer, config=deepcopy(config)
+    )
+
+
+def _make_direct_config() -> HybridModelConfig:
+    transformer = _make_transformer()
+    layer_specs = [
+        [_make_layer(transformer)],
+        PipelineSplit(),
+        [_make_layer(transformer)],
+        PipelineSplit(),
+        [_make_layer(transformer)],
+        PipelineSplit(),
+        [_make_layer(transformer)],
+    ]
+    return HybridModelConfig(transformer=transformer, vocab_size=32000, layer_specs=layer_specs)
+
+
+class TestDirectHybridModelConfig:
+    def test_direct_fields_default_to_none(self):
+        config = HybridModelConfig(transformer=_make_transformer())
+
+        assert config.layer_specs is None
+        assert config.mtp_layer_specs is None
+
+    def test_direct_fields_preserve_python_architecture_objects(self):
+        transformer = _make_transformer()
+        layer_specs = [_make_layer(transformer)]
+        mtp_layer_specs = [_make_layer(transformer)]
+
+        config = HybridModelConfig(
+            transformer=transformer, layer_specs=layer_specs, mtp_layer_specs=mtp_layer_specs
+        )
+
+        assert config.layer_specs is layer_specs
+        assert config.mtp_layer_specs is mtp_layer_specs
+
+    def test_as_dict_excludes_raw_direct_specs(self):
+        transformer = _make_transformer()
+        config = HybridModelConfig(
+            transformer=transformer,
+            vocab_size=32000,
+            layer_specs=[_make_layer(transformer)],
+            mtp_layer_specs=[_make_layer(transformer)],
+        )
+
+        serialized = config.as_dict()
+
+        assert "layer_specs" not in serialized
+        assert "mtp_layer_specs" not in serialized
+        assert serialized["vocab_size"] == 32000
+        assert serialized["transformer"]["num_layers"] == 4
+
+    def test_checkpoint_args_exclude_runtime_resolved_architecture(self):
+        args = SimpleNamespace(
+            resolved_hybrid_architecture=SimpleNamespace(source="direct"),
+            no_save_optim=True,
+            no_save_rng=True,
+        )
+
+        state_dict = generate_state_dict(
+            args, model=[], optimizer=None, opt_param_scheduler=None, rng_state=None
+        )
+
+        assert hasattr(args, "resolved_hybrid_architecture")
+        assert not hasattr(state_dict["args"], "resolved_hybrid_architecture")
+
+    def test_legacy_checkpoint_keeps_the_original_args_object(self):
+        args = SimpleNamespace(
+            hybrid_layer_pattern="M*",
+            resolved_hybrid_architecture=SimpleNamespace(source="legacy"),
+            no_save_optim=True,
+            no_save_rng=True,
+        )
+
+        state_dict = generate_state_dict(
+            args, model=[], optimizer=None, opt_param_scheduler=None, rng_state=None
+        )
+
+        assert state_dict["args"] is args
+        assert state_dict["args"].resolved_hybrid_architecture.source == "legacy"
+
+
+class TestDirectHybridModelBuilder:
+    def test_prepares_inferred_vpp_before_distributed_initialization(self):
+        config = _make_direct_config()
+        args = SimpleNamespace(
+            pipeline_model_parallel_size=2,
+            virtual_pipeline_model_parallel_size=None,
+            overlap_p2p_comm=False,
+            batch_p2p_comm=True,
+            align_param_gather=False,
+            _overlap_p2p_comm_before_direct_vpp=True,
+            _align_param_gather_before_direct_vpp=True,
+        )
+
+        prepared = HybridModelBuilder.prepare_config_for_distributed_init(config, args)
+
+        assert prepared is True
+        assert args.virtual_pipeline_model_parallel_size == 2
+        assert args.overlap_p2p_comm is True
+        assert args.batch_p2p_comm is False
+        assert args.align_param_gather is True
+        assert config.transformer.virtual_pipeline_model_parallel_size == 2
+        assert config.transformer.overlap_p2p_comm is True
+        assert config.transformer.batch_p2p_comm is False
+
+    def test_cli_validation_defers_uneven_direct_pp_vpp_topology(self, monkeypatch):
+        monkeypatch.setattr(sys, "argv", ["test_hybrid_direct_builder.py"])
+        monkeypatch.setattr("megatron.training.arguments._print_args", lambda *args: None)
+        args = parse_args()
+        args.world_size = 2
+        args.rank = 0
+        args.pipeline_model_parallel_size = 2
+        args.num_layers = 5
+        args.hidden_size = 128
+        args.num_attention_heads = 4
+        args.max_position_embeddings = 128
+        args.seq_length = 128
+        args.micro_batch_size = 1
+        args.train_iters = 1
+        args.lr = 1.0e-4
+        args.tokenizer_type = "NullTokenizer"
+        args.vocab_size = 1024
+
+        validate_args(args)
+        assert args.virtual_pipeline_model_parallel_size is None
+        assert args.overlap_p2p_comm is False
+        assert args.align_param_gather is False
+
+        transformer = core_transformer_config_from_args(args)
+        layer = _make_layer(transformer)
+        model_config = HybridModelConfig(
+            transformer=transformer,
+            vocab_size=1024,
+            layer_specs=[
+                layer,
+                PipelineSplit(),
+                [],
+                PipelineSplit(),
+                [],
+                PipelineSplit(),
+                [layer] * 4,
+            ],
+        )
+
+        prepared = HybridModelBuilder.prepare_config_for_distributed_init(model_config, args)
+
+        assert prepared is True
+        assert args.virtual_pipeline_model_parallel_size == 2
+        assert args.overlap_p2p_comm is True
+        assert args.align_param_gather is True
+        assert transformer.overlap_p2p_comm is True
+        assert transformer.batch_p2p_comm is False
+
+    def test_cli_validation_does_not_add_direct_vpp_state_to_legacy_patterns(self, monkeypatch):
+        monkeypatch.setattr(sys, "argv", ["test_hybrid_direct_builder.py"])
+        monkeypatch.setattr("megatron.training.arguments._print_args", lambda *args: None)
+        args = parse_args()
+        args.world_size = 2
+        args.rank = 0
+        args.pipeline_model_parallel_size = 2
+        args.num_layers = 2
+        args.hidden_size = 128
+        args.num_attention_heads = 4
+        args.max_position_embeddings = 128
+        args.seq_length = 128
+        args.micro_batch_size = 1
+        args.train_iters = 1
+        args.lr = 1.0e-4
+        args.tokenizer_type = "NullTokenizer"
+        args.vocab_size = 1024
+        args.hybrid_layer_pattern = "M*"
+
+        validate_args(args)
+
+        assert not hasattr(args, "_overlap_p2p_comm_before_direct_vpp")
+        assert not hasattr(args, "_align_param_gather_before_direct_vpp")
+
+    @pytest.mark.parametrize(
+        ("runtime_pp_size", "runtime_vp_size", "message"),
+        [
+            pytest.param(1, None, "pipeline_model_parallel_size must match", id="pp"),
+            pytest.param(2, 3, "splits disagree", id="vpp"),
+        ],
+    )
+    def test_pre_init_topology_must_match_runtime(self, runtime_pp_size, runtime_vp_size, message):
+        config = _make_direct_config()
+        args = SimpleNamespace(
+            pipeline_model_parallel_size=runtime_pp_size,
+            virtual_pipeline_model_parallel_size=runtime_vp_size,
+            overlap_p2p_comm=True,
+        )
+
+        with pytest.raises(ValueError, match=message):
+            HybridModelBuilder.prepare_config_for_distributed_init(config, args)
+
+    @patch("megatron.training.models.hybrid.compose_hooks")
+    @patch("megatron.training.models.hybrid.unimodal_build_distributed_models")
+    def test_resolves_before_distributed_construction_and_infers_vpp(
+        self, mock_unimodal, mock_compose
+    ):
+        config = _make_direct_config()
+        assert config.transformer.virtual_pipeline_model_parallel_size is None
+
+        builder = HybridModelBuilder(config)
+        assert config.transformer.virtual_pipeline_model_parallel_size == 2
+        assert [len(segment) for segment in builder._resolved_architecture.segments] == [1, 1, 1, 1]
+
+        observed_vp_sizes = []
+
+        def fake_distributed_builder(*args):
+            observed_vp_sizes.append(args[1].virtual_pipeline_model_parallel_size)
+            return []
+
+        mock_unimodal.side_effect = fake_distributed_builder
+        mock_compose.return_value = Mock(return_value=None)
+
+        builder.build_distributed_models(Mock(), wrap_with_ddp=False)
+
+        assert observed_vp_sizes == [2]
+
+    @pytest.mark.parametrize(
+        ("pp_rank", "vp_stage", "expected_pre", "expected_post"),
+        [(0, 0, True, False), (1, 0, False, False), (0, 1, False, False), (1, 1, False, True)],
+        ids=["first-pp-first-vp", "last-pp-first-vp", "first-pp-last-vp", "last-pp-last-vp"],
+    )
+    @patch("megatron.training.models.hybrid.HybridModel")
+    def test_default_pre_post_process_ownership(
+        self, mock_model, pp_rank, vp_stage, expected_pre, expected_post
+    ):
+        builder = HybridModelBuilder(_make_direct_config())
+        pg_collection = Mock()
+        pg_collection.pp = Mock()
+
+        with (
+            patch("megatron.training.models.hybrid.is_pp_first_stage", return_value=pp_rank == 0),
+            patch("megatron.training.models.hybrid.is_pp_last_stage", return_value=pp_rank == 1),
+        ):
+            builder.build_model(pg_collection, vp_stage=vp_stage)
+
+        kwargs = mock_model.call_args.kwargs
+        assert kwargs["pre_process"] is expected_pre
+        assert kwargs["post_process"] is expected_post
+
+    @patch("megatron.training.models.hybrid.HybridModel")
+    def test_passes_resolved_architecture_to_hybrid_model(self, mock_model):
+        builder = HybridModelBuilder(_make_direct_config())
+        pg_collection = Mock()
+
+        builder.build_model(pg_collection, pre_process=False, post_process=False, vp_stage=1)
+
+        kwargs = mock_model.call_args.kwargs
+        assert builder._resolved_architecture is not None
+        assert kwargs["resolved_hybrid_architecture"] is builder._resolved_architecture
+        assert kwargs["hybrid_layer_pattern"] is None
+        assert kwargs["vp_stage"] == 1
+
+    @patch("megatron.training.models.hybrid.is_pp_first_stage", return_value=True)
+    @patch("megatron.training.models.hybrid.is_pp_last_stage", return_value=False)
+    @patch("megatron.training.models.hybrid.HybridModel")
+    def test_single_chunk_build_defaults_to_first_virtual_stage(
+        self, mock_model, _mock_last_stage, _mock_first_stage
+    ):
+        builder = HybridModelBuilder(_make_direct_config())
+
+        builder.build_model(Mock())
+
+        kwargs = mock_model.call_args.kwargs
+        assert kwargs["vp_stage"] == 0
+        assert kwargs["pre_process"] is True
+        assert kwargs["post_process"] is False