diff --git a/megatron/core/models/hybrid/hybrid_block.py b/megatron/core/models/hybrid/hybrid_block.py
index 93cb56f5297..bfdeb6029f1 100644
--- a/megatron/core/models/hybrid/hybrid_block.py
+++ b/megatron/core/models/hybrid/hybrid_block.py
@@ -7,7 +7,7 @@
 
 from contextlib import nullcontext
 from dataclasses import dataclass
-from typing import Optional, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 import torch
 from torch import Tensor, nn
@@ -22,12 +22,21 @@
 from megatron.core.models.hybrid.hybrid_layer_allocation import Symbols as LayerSymbols
 from megatron.core.packed_seq_params import PackedSeqParams
 from megatron.core.process_groups_config import ProcessGroupCollection
+from megatron.core.tensor_parallel.random import CheckpointManager
 from megatron.core.transformer import TransformerConfig
+from megatron.core.transformer.hyper_connection import (
+    HyperConnectionModule,
+    learned_output_contract,
+)
 from megatron.core.transformer.identity_op import IdentityOp
 from megatron.core.transformer.module import MegatronModule
 from megatron.core.transformer.spec_utils import ModuleSpec, build_module
 from megatron.core.transformer.transformer_layer import TransformerLayer
-from megatron.core.transformer.utils import sharded_state_dict_default
+from megatron.core.transformer.utils import (
+    ensure_metadata_has_dp_cp_group,
+    make_sharded_tensors_for_checkpoint,
+    sharded_state_dict_default,
+)
 from megatron.core.utils import WrappedTensor, deprecate_inference_params, make_viewless_tensor
 
 
@@ -46,6 +55,165 @@ class HybridStackSubmodules:
     mtp_block_spec: Optional[ModuleSpec] = None
 
 
+class HyperConnectionHybridLayer(MegatronModule):
+    """Layer-boundary mHC wrapper for HybridStack layers.
+
+    Hybrid layers already own their local residual paths. For this initial
+    integration we treat each hybrid layer as a single function by aggregating
+    n streams to the layer input, running the existing layer, and feeding only
+    the layer delta back through mHC expansion. The expansion path intentionally
+    uses zero additional dropout because the wrapped hybrid layer has already
+    applied its local dropout/residual update before the delta is computed.
+
+    Checkpoint compatibility: this is a *wrapper* (the inner layer is held as
+    `self.inner_layer`), so wrapped-layer state_dict keys are nested under
+    `inner_layer.` (e.g. `layers.0.inner_layer.input_layernorm.weight` instead
+    of `layers.0.input_layernorm.weight`). HybridStack checkpoints saved with
+    `enable_hyper_connections=False` cannot be loaded into a model with
+    `enable_hyper_connections=True` (and vice versa) without a key-mapping
+    migration. Note: this differs from `HyperConnectionTransformerLayer`,
+    which subclasses `TransformerLayer` and only adds new sibling fields,
+    keeping all base keys stable.
+    """
+
+    def __init__(self, config: TransformerConfig, layer: MegatronModule) -> None:
+        super().__init__(config=config)
+        self.inner_layer = layer
+        self.layer_number = layer.layer_number
+        self.hyper_connection = HyperConnectionModule(config=config, layer_number=self.layer_number)
+        if config.params_dtype is not None:
+            self.hyper_connection.to(dtype=config.params_dtype)
+        if hasattr(layer, 'tp_group'):
+            self.tp_group = layer.tp_group
+
+    def mamba_state_shapes_per_request(self) -> Optional[Tuple[Tuple[int], Tuple[int]]]:
+        """Delegate Mamba inference state shape requests to the wrapped layer."""
+        if not hasattr(self.inner_layer, 'mamba_state_shapes_per_request'):
+            return None
+        return self.inner_layer.mamba_state_shapes_per_request()
+
+    def _call_inner_layer(
+        self,
+        hidden_states: Tensor,
+        attention_mask: Tensor,
+        inference_context: Optional[BaseInferenceContext],
+        rotary_pos_emb: Optional[Tensor],
+        sequence_len_offset: Optional[Tensor],
+        packed_seq_params: Optional[PackedSeqParams],
+        padding_mask: Optional[Tensor],
+    ) -> Tuple[Tensor, Optional[Tensor]]:
+        if isinstance(self.inner_layer, TransformerLayer):
+            output = self.inner_layer(
+                hidden_states=hidden_states,
+                attention_mask=attention_mask,
+                inference_context=inference_context,
+                rotary_pos_emb=rotary_pos_emb,
+                sequence_len_offset=sequence_len_offset,
+                packed_seq_params=packed_seq_params,
+                padding_mask=padding_mask,
+                _called_from_hybrid_mhc_wrapper=True,
+            )
+        else:
+            # Non-transformer layers (e.g. MambaLayer; GatedDeltaNet which does
+            # accept `sequence_len_offset` is currently always wrapped inside a
+            # TransformerLayer spec, so it takes the branch above) do not accept
+            # rotary_pos_emb / sequence_len_offset / padding_mask — pass only
+            # the common arguments. New layer types that consume any of these
+            # must add explicit handling here.
+            output = self.inner_layer(
+                hidden_states=hidden_states,
+                attention_mask=attention_mask,
+                inference_context=inference_context,
+                packed_seq_params=packed_seq_params,
+            )
+
+        if isinstance(output, tuple):
+            context = output[1] if len(output) > 1 else None
+            return output[0], context
+        return output, None
+
+    def forward(
+        self,
+        hidden_states: Tensor,
+        attention_mask: Tensor,
+        inference_context: Optional[BaseInferenceContext] = None,
+        rotary_pos_emb: Optional[Tensor] = None,
+        sequence_len_offset: Optional[Tensor] = None,
+        packed_seq_params: Optional[PackedSeqParams] = None,
+        padding_mask: Optional[Tensor] = None,
+        mhc_recompute_manager=None,
+    ) -> Tuple[Tensor, Optional[Tensor]]:
+        """Run the wrapped hybrid layer through one layer-boundary mHC update."""
+        residual = hidden_states
+        aggregated, h_res, h_post = self.hyper_connection(
+            hidden_states, mhc_recompute_manager=mhc_recompute_manager
+        )
+        layer_output, context = self._call_inner_layer(
+            aggregated,
+            attention_mask,
+            inference_context,
+            rotary_pos_emb,
+            sequence_len_offset,
+            packed_seq_params,
+            padding_mask,
+        )
+        # The inner hybrid layer already applied its own local residual/dropout, so
+        # it returns `aggregated + f(aggregated)`. We feed only the function
+        # delta `f(aggregated)` into the n-stream BDA so it does not double-count
+        # the residual that mHC owns. The temporary [s, b, C] tensor here is the
+        # simplest correct form; a future optimization could fuse the subtraction
+        # into `fused_h_res_h_post_bda` to avoid the allocation.
+        # Sanity check: this contract requires the inner layer to preserve shape;
+        # any mismatch indicates a future layer type is breaking the residual
+        # assumption and would silently corrupt the n-stream state.
+        if layer_output.shape != aggregated.shape:
+            raise RuntimeError(
+                "HyperConnectionHybridLayer requires inner layers to preserve "
+                f"hidden-state shape. Got {tuple(layer_output.shape)} from inner layer "
+                f"vs {tuple(aggregated.shape)} input; layer must add its own residual."
+            )
+        # `fp32_residual_connection=True` may cause some inner layers (e.g.,
+        # MambaLayer) to return `layer_output` in fp32 while `aggregated` is in
+        # compute dtype; explicitly upcast `aggregated` so the subtraction stays
+        # in fp32 instead of relying on PyTorch's implicit promotion.
+        if self.config.fp32_residual_connection and aggregated.dtype != layer_output.dtype:
+            aggregated = aggregated.to(layer_output.dtype)
+        layer_delta = layer_output - aggregated
+        # `dropout_prob=0.0` already disables dropout regardless of training mode;
+        # `training=self.training` is more semantically accurate than hard-coding
+        # False during a training-mode forward.
+        is_last_in_recompute_block = bool(
+            mhc_recompute_manager is not None
+            and getattr(mhc_recompute_manager, "is_last_layer_in_recompute_block", False)
+        )
+        mhc_bda_manager = None if is_last_in_recompute_block else mhc_recompute_manager
+
+        hidden_states = self.hyper_connection.fused_h_res_h_post_bda(
+            h_res,
+            residual,
+            h_post,
+            (layer_delta, None),
+            dropout_prob=0.0,
+            training=self.training,
+            fused=False,
+            manager=mhc_bda_manager,
+        )
+        # In `HyperConnectionTransformerLayer` the n-stream output stays in compute
+        # dtype because the post-attention `x` is in compute dtype. In the hybrid
+        # wrapper, `layer_delta` may be fp32 (when `fp32_residual_connection=True`
+        # or an inner layer upcasts), so `fused_h_res_h_post_bda`'s `output.to(x.dtype)`
+        # would leave the result in fp32 and silently propagate fp32 n-stream
+        # hidden states to every subsequent layer (~2x activation memory). Restore
+        # the compute-dtype contract here.
+        if (
+            self.config.fp32_residual_connection
+            and self.config.params_dtype is not None
+            and hidden_states.dtype != self.config.params_dtype
+        ):
+            hidden_states = hidden_states.to(self.config.params_dtype)
+        return hidden_states, context
+
+
 class HybridStack(MegatronModule):
     """
     Constructor for the HybridStack class.
@@ -100,6 +268,10 @@ def __init__(
         self.input_tensor = None
         self.pg_collection = pg_collection
 
+        # Lazily populated mHC recompute layout cache (deterministic from config
+        # and num_layers); see `_build_mhc_recompute_layer_plan`.
+        self._mhc_block_end_plan: Optional[List[bool]] = None
+
         assert layer_type_list is not None, (
             "layer_type_list must be provided. It should be pre-computed from "
             "--hybrid-layer-pattern by HybridModel."
@@ -172,6 +344,8 @@ def __init__(
                     )
                 else:
                     raise ValueError("unexpected layer_type")
+            if self.config.enable_hyper_connections:
+                layer = HyperConnectionHybridLayer(config=self.config, layer=layer)
             self.layers.append(layer)
 
         # Required for activation recomputation
@@ -185,6 +359,18 @@ def __init__(
                 eps=self.config.layernorm_epsilon,
             )
 
+        if self.config.enable_hyper_connections and self.post_process:
+            hc_mult = self.config.num_residual_streams
+            hc_dim = self.config.hidden_size * hc_mult
+            self.hc_head_fn = nn.Parameter(torch.randn(hc_mult, hc_dim))
+            self.hc_head_base = nn.Parameter(torch.zeros(hc_mult))
+            self.hc_head_scale = nn.Parameter(torch.ones(1))
+            nn.init.xavier_uniform_(self.hc_head_fn)
+            if self.config.sequence_parallel:
+                setattr(self.hc_head_fn, 'sequence_parallel', True)
+                setattr(self.hc_head_base, 'sequence_parallel', True)
+                setattr(self.hc_head_scale, 'sequence_parallel', True)
+
     def set_input_tensor(self, input_tensor: Tensor):
         """Set input tensor to be used instead of forward()'s input.
 
@@ -205,6 +391,59 @@ def mamba_state_shapes_per_request(self) -> Optional[Tuple[Tuple[int], Tuple[int
                 return layer.mamba_state_shapes_per_request()
         return None
 
+    def _compute_mhc_block_end_plan(self) -> List[bool]:
+        """Compute per-layer block-end markers (deterministic from config)."""
+        num_layers = len(self.layers)
+        is_recompute_block_end: List[bool] = [False] * num_layers
+        if num_layers == 0:
+            return is_recompute_block_end
+        mhc_recompute_layer_num = self.config.mhc_recompute_layer_num
+        for l_no in range(num_layers):
+            is_last_in_stack = l_no == num_layers - 1
+            is_last_in_recompute_block = is_last_in_stack
+            if mhc_recompute_layer_num is not None:
+                is_last_in_recompute_block = is_last_in_stack or (
+                    (l_no + 1) % mhc_recompute_layer_num == 0
+                )
+            is_recompute_block_end[l_no] = is_last_in_recompute_block
+        return is_recompute_block_end
+
+    def _build_mhc_recompute_layer_plan(
+        self, use_mhc_recompute: bool
+    ) -> Tuple[List[Optional[CheckpointManager]], List[bool]]:
+        """Pre-build per-layer MHC recompute managers and block-end markers.
+
+        The block-end plan is deterministic from config and cached on the
+        instance; only the per-block ``CheckpointManager`` instances are
+        allocated fresh per forward pass (managers are single-use). Mirrors
+        the caching scheme used by ``TransformerBlock``.
+        """
+        num_layers = len(self.layers)
+        if not use_mhc_recompute or num_layers == 0:
+            return [None] * num_layers, [False] * num_layers
+
+        if self._mhc_block_end_plan is None:
+            self._mhc_block_end_plan = self._compute_mhc_block_end_plan()
+        is_recompute_block_end = self._mhc_block_end_plan
+
+        layer_managers: List[Optional[CheckpointManager]] = [None] * num_layers
+        mhc_manager = CheckpointManager()
+        for l_no in range(num_layers):
+            layer_managers[l_no] = mhc_manager
+            if is_recompute_block_end[l_no] and l_no != num_layers - 1:
+                mhc_manager = CheckpointManager()
+        return layer_managers, is_recompute_block_end
+
+    @staticmethod
+    def _finalize_mhc_recompute_layer(
+        mhc_manager: Optional[CheckpointManager],
+        hidden_states: Tensor,
+        is_last_in_recompute_block: bool,
+    ) -> None:
+        """Finalize MHC recompute state for the current layer when a block ends."""
+        if mhc_manager is not None and is_last_in_recompute_block:
+            mhc_manager.discard_all_outputs_and_register_unified_recompute(hidden_states)
+
     def forward(
         self,
         hidden_states: Union[Tensor, WrappedTensor],
@@ -244,6 +483,11 @@ def forward(
         if isinstance(hidden_states, WrappedTensor):
             hidden_states = hidden_states.unwrap()
 
+        if self.config.enable_hyper_connections and self.pre_process:
+            hidden_states = HyperConnectionModule.input_expand(
+                hidden_states, self.config.num_residual_streams
+            )
+
         if inference_context and inference_context.is_static_batching():
             # NOTE(bnorick): match BaseInferenceContext attributes for
             # mamba_ssm.utils.generation.BaseInferenceContext,
@@ -291,13 +535,29 @@ def get_inner_quant_context(config, layer_number):
             def get_inner_quant_context(config, layer_number):
                 return nullcontext()
 
+        use_mhc_recompute = (
+            self.training
+            and self.config.enable_hyper_connections
+            and self.config.recompute_granularity == 'selective'
+            and "mhc" in self.config.recompute_modules
+        )
+        mhc_layer_managers, mhc_is_last_in_recompute_block = self._build_mhc_recompute_layer_plan(
+            use_mhc_recompute
+        )
+
         with outer_fp8_context:
-            for layer in self.layers:
+            for l_no, layer in enumerate(self.layers):
                 # Layers have 1-indexed layer numbers attribute.
                 inner_quant_context = get_inner_quant_context(self.config, layer.layer_number - 1)
+                mhc_manager = mhc_layer_managers[l_no]
+                if mhc_manager is not None:
+                    mhc_manager.is_last_layer_in_recompute_block = mhc_is_last_in_recompute_block[
+                        l_no
+                    ]
+
                 with inner_quant_context:
-                    if isinstance(layer, TransformerLayer):
-                        hidden_states, _ = layer(
+                    if isinstance(layer, (TransformerLayer, HyperConnectionHybridLayer)):
+                        layer_kwargs = dict(
                             hidden_states=hidden_states,
                             attention_mask=attention_mask,
                             inference_context=inference_context,
@@ -306,6 +566,11 @@ def get_inner_quant_context(config, layer_number):
                             packed_seq_params=packed_seq_params,
                             padding_mask=padding_mask,
                         )
+                        if mhc_manager is not None and isinstance(
+                            layer, HyperConnectionHybridLayer
+                        ):
+                            layer_kwargs["mhc_recompute_manager"] = mhc_manager
+                        hidden_states, _ = layer(**layer_kwargs)
                     else:  # MambaLayer, Expert, or MLP
                         hidden_states = layer(
                             hidden_states=hidden_states,
@@ -320,6 +585,22 @@ def get_inner_quant_context(config, layer_number):
                 if isinstance(hidden_states, tuple):
                     hidden_states = hidden_states[0]
 
+                self._finalize_mhc_recompute_layer(
+                    mhc_manager=mhc_manager,
+                    hidden_states=hidden_states,
+                    is_last_in_recompute_block=mhc_is_last_in_recompute_block[l_no],
+                )
+
+        if self.config.enable_hyper_connections and self.post_process:
+            hidden_states = learned_output_contract(
+                hidden_states,
+                self.hc_head_fn,
+                self.hc_head_base,
+                self.hc_head_scale,
+                self.config.num_residual_streams,
+                self.config.layernorm_epsilon,
+            )
+
         # Final layer norm.
         if self.post_process and self.post_layer_norm:
             hidden_states = self.final_norm(hidden_states)
@@ -354,6 +635,7 @@ def sharded_state_dict(
             dict: The sharded state dictionary for the current object.
         """
 
+        sharded_offsets = sharded_offsets or ()
         sharded_state_dict = {}
         layer_prefix = f'{prefix}layers.'
 
@@ -388,6 +670,20 @@ def sharded_state_dict(
                     )
                 )
 
+        local_state_dict: dict = {}
+        self._save_to_state_dict(local_state_dict, '', keep_vars=True)
+        if local_state_dict:
+            metadata = ensure_metadata_has_dp_cp_group(metadata)
+            sharded_state_dict.update(
+                make_sharded_tensors_for_checkpoint(
+                    local_state_dict,
+                    prefix,
+                    sharded_offsets=sharded_offsets or (),
+                    tp_group=self.tp_group,
+                    dp_cp_group=metadata['dp_cp_group'],
+                )
+            )
+
         return sharded_state_dict
 
 
diff --git a/megatron/core/transformer/transformer_layer.py b/megatron/core/transformer/transformer_layer.py
index d2e090de232..353979f97f9 100644
--- a/megatron/core/transformer/transformer_layer.py
+++ b/megatron/core/transformer/transformer_layer.py
@@ -716,9 +716,15 @@ def forward(self, *args, **kwargs):
         """
         # Injected by __call__ for cuda graph keying; not a real forward arg.
         kwargs.pop("dynamic_inference_decode_only", None)
-        assert (
-            not self.config.enable_hyper_connections
-        ), "Please use HyperConnectionTransformerLayer instead"
+        called_from_hybrid_mhc_wrapper = kwargs.pop("_called_from_hybrid_mhc_wrapper", False)
+        if self.config.enable_hyper_connections and not called_from_hybrid_mhc_wrapper:
+            raise RuntimeError(
+                "TransformerLayer.forward() must not be called directly when "
+                "enable_hyper_connections=True. Use HyperConnectionTransformerLayer "
+                "for transformer-only stacks; HyperConnectionHybridLayer drives the "
+                "wrapped TransformerLayer through this path automatically for hybrid "
+                "stacks."
+            )
         hidden_states, context = self._forward_attention(*args, **kwargs)
         output = self._forward_mlp(
             hidden_states,
@@ -1604,6 +1610,7 @@ def _get_submodules_under_cudagraphs(self):
     def forward(self, *args, **kwargs):
         """Forward pass with MHC recompute manager support."""
         kwargs.pop("dynamic_inference_decode_only", None)
+        kwargs.pop("_called_from_hybrid_mhc_wrapper", None)
 
         mhc_recompute_manager = getattr(self, '_mhc_recompute_manager', None)
 
diff --git a/tests/unit_tests/models/test_dsa_gpt_mamba_equivalence.py b/tests/unit_tests/models/test_dsa_gpt_mamba_equivalence.py
index 229af268a79..9255e4794d5 100644
--- a/tests/unit_tests/models/test_dsa_gpt_mamba_equivalence.py
+++ b/tests/unit_tests/models/test_dsa_gpt_mamba_equivalence.py
@@ -35,6 +35,7 @@
     get_transformer_block_with_experimental_attention_variant_spec,
 )
 from megatron.core.models.gpt.gpt_model import GPTModel
+from megatron.core.models.hybrid.hybrid_block import HyperConnectionHybridLayer
 from megatron.core.models.hybrid.hybrid_layer_allocation import validate_segment_layers
 from megatron.core.models.hybrid.hybrid_layer_specs import hybrid_stack_spec
 from megatron.core.models.hybrid.hybrid_model import HybridModel
@@ -642,3 +643,52 @@ def test_moe_record_and_compare_golden_values(self, tp: int, pp: int) -> None:
 
         # Verify HybridModel matches golden values
         _compare_against_golden_values(mamba_logprobs, gpt_logprobs, abs_tol=1e-3)
+
+
+# ---------------------------------------------------------------------------
+# mHC HybridModel smoke tests for DeepSeek proxy patterns
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+class TestDSAHybridMHCProxy:
+    """Smoke-test mHC on DeepSeek-style HybridModel patterns.
+
+    These do not assert GPT/Hybrid numerical equivalence because the current
+    HybridModel implementation wraps each split hybrid layer at the boundary,
+    whereas GPT mHC has separate attention and MLP hyper-connections inside a
+    TransformerLayer.
+    """
+
+    def teardown_method(self, method):
+        Utils.destroy_model_parallel()
+
+    def _enable_mhc(self, config: MLATransformerConfig) -> MLATransformerConfig:
+        config.enable_hyper_connections = True
+        config.num_residual_streams = 4
+        config.mhc_sinkhorn_iterations = 5
+        config.mhc_init_gating_factor = 0.01
+        config.hidden_dropout = 0.0
+        return config
+
+    def _assert_mhc_model_forward(self, config: MLATransformerConfig, pattern: str) -> None:
+        Utils.initialize_model_parallel(1, 1)
+        model_parallel_cuda_manual_seed(42)
+        model = _build_mamba_model(self._enable_mhc(config), pattern)
+        assert all(isinstance(layer, HyperConnectionHybridLayer) for layer in model.decoder.layers)
+
+        torch.manual_seed(99)
+        tokens = torch.randint(0, _VOCAB_SIZE, (_BATCH_SIZE, _SEQ_LEN), device='cuda')
+        logprobs = _forward_logprobs_pp1(model, tokens)
+        assert logprobs.shape == (_BATCH_SIZE, _SEQ_LEN - 1)
+        assert torch.isfinite(logprobs).all()
+
+    def test_dsa_dense_hybrid_mhc_forward(self) -> None:
+        """DeepSeek-V3.2-style DSA + MLP split pattern runs with mHC."""
+        config = _make_dsa_config(num_layers=_NUM_GPT_LAYERS, tp=1, pp=1)
+        self._assert_mhc_model_forward(config, _MAMBA_PATTERN)
+
+    def test_dsa_moe_hybrid_mhc_forward(self) -> None:
+        """DeepSeek-V3-style DSA + MoE split pattern runs with mHC."""
+        config = _make_dsa_moe_config(num_layers=_NUM_GPT_LAYERS, tp=1, pp=1)
+        self._assert_mhc_model_forward(config, _MOE_MAMBA_PATTERN)
diff --git a/tests/unit_tests/models/test_hybrid_model.py b/tests/unit_tests/models/test_hybrid_model.py
index 98a53da0314..d17d244fa8f 100644
--- a/tests/unit_tests/models/test_hybrid_model.py
+++ b/tests/unit_tests/models/test_hybrid_model.py
@@ -16,17 +16,61 @@
 from megatron.core.inference.inference_request import DynamicInferenceRequest
 from megatron.core.inference.sampling_params import SamplingParams
 from megatron.core.models.common.embeddings.yarn_rotary_pos_embedding import YarnRotaryEmbedding
+from megatron.core.models.hybrid.hybrid_block import (
+    HybridStack,
+    HybridStackSubmodules,
+    HyperConnectionHybridLayer,
+)
 from megatron.core.models.hybrid.hybrid_layer_specs import hybrid_stack_spec
 from megatron.core.models.hybrid.hybrid_model import HybridModel
 from megatron.core.packed_seq_params import PackedSeqParams
 from megatron.core.tensor_parallel.random import model_parallel_cuda_manual_seed
 from megatron.core.transformer import TransformerConfig
 from megatron.core.transformer.enums import AttnBackend
-from megatron.core.transformer.module import Float16Module
+from megatron.core.transformer.module import Float16Module, MegatronModule
+from megatron.core.transformer.spec_utils import ModuleSpec
 from megatron.core.utils import divide, is_fa_min_version, is_torch_min_version
 from tests.unit_tests.test_utilities import Utils
 
 
+class _DummyHybridLayer(MegatronModule):
+    """Minimal same-shape layer used to test HybridModel/mHC plumbing."""
+
+    def __init__(self, config: TransformerConfig, layer_number: int, **_kwargs):
+        super().__init__(config=config)
+        self.layer_number = layer_number
+        self.proj = torch.nn.Linear(config.hidden_size, config.hidden_size, bias=False)
+        self.seen_hidden_shapes = []
+
+    def forward(
+        self,
+        hidden_states,
+        attention_mask=None,
+        inference_context=None,
+        packed_seq_params=None,
+        **_kwargs,
+    ):
+        self.seen_hidden_shapes.append(tuple(hidden_states.shape))
+        return hidden_states + 0.125 * self.proj(hidden_states)
+
+
+def _get_dummy_hybrid_stack_spec() -> ModuleSpec:
+    """Build a HybridStack spec whose layer symbols all resolve to dummy layers."""
+    dummy_layer_spec = ModuleSpec(module=_DummyHybridLayer)
+    return ModuleSpec(
+        module=HybridStack,
+        params={"post_layer_norm": False},
+        submodules=HybridStackSubmodules(
+            mamba_layer=dummy_layer_spec,
+            gdn_layer=dummy_layer_spec,
+            attention_layer=dummy_layer_spec,
+            dsa_layer=dummy_layer_spec,
+            mlp_layer=dummy_layer_spec,
+            moe_layer=dummy_layer_spec,
+        ),
+    )
+
+
 class TestHybridModel:
 
     def setup_method(self, method):
@@ -57,6 +101,176 @@ def test_constructor(self):
         num_weights = sum([p.numel() for p in self.model.parameters()])
         assert num_weights == 1774872
 
+    def test_constructor_with_hyper_connections(self):
+        model_config = TransformerConfig(
+            num_layers=3,
+            hidden_size=256,
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+        )
+        model = HybridModel(
+            config=model_config,
+            hybrid_stack_spec=hybrid_stack_spec,
+            vocab_size=100,
+            max_sequence_length=4,
+            hybrid_layer_pattern="M*-",
+        )
+
+        assert all(isinstance(layer, HyperConnectionHybridLayer) for layer in model.decoder.layers)
+        assert model.decoder.hc_head_fn.shape == (
+            model_config.num_residual_streams,
+            model_config.hidden_size * model_config.num_residual_streams,
+        )
+        assert model.decoder.hc_head_base.shape == (model_config.num_residual_streams,)
+        assert model.decoder.hc_head_scale.shape == (1,)
+        assert "decoder.hc_head_fn" in model.state_dict()
+        decoder_sharded_state = model.decoder.sharded_state_dict(prefix="decoder.", metadata={})
+        assert "decoder.hc_head_fn" in decoder_sharded_state
+        assert "decoder.hc_head_base" in decoder_sharded_state
+        assert "decoder.hc_head_scale" in decoder_sharded_state
+        num_weights = sum([p.numel() for p in model.parameters()])
+        assert num_weights > sum([p.numel() for p in self.model.parameters()])
+
+    def test_hyper_connection_recompute_skips_boundary_bda_checkpoint(self, monkeypatch):
+        model_config = TransformerConfig(
+            num_layers=1,
+            hidden_size=8,
+            num_attention_heads=1,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+            mhc_sinkhorn_iterations=3,
+        )
+        layer = HyperConnectionHybridLayer(
+            config=model_config, layer=_DummyHybridLayer(model_config, layer_number=1)
+        )
+        hidden_states = torch.randn(
+            4, 2, model_config.hidden_size * model_config.num_residual_streams, requires_grad=True
+        )
+        manager = type("_FakeManager", (), {})()
+        manager.is_last_layer_in_recompute_block = True
+        seen_bda_managers = []
+
+        def fake_hyper_connection_forward(hidden_states, mhc_recompute_manager=None):
+            assert mhc_recompute_manager is manager
+            s, b, _ = hidden_states.shape
+            n = model_config.num_residual_streams
+            c = model_config.hidden_size
+            aggregated = hidden_states.view(s, b, n, c).mean(dim=2)
+            h_res = torch.empty(s, b, n, n, dtype=hidden_states.dtype)
+            h_post = torch.empty(s, b, n, dtype=hidden_states.dtype)
+            return aggregated, h_res, h_post
+
+        def fake_fused_h_res_h_post_bda(
+            h_res,
+            original_residual,
+            h_post,
+            layer_output_with_bias,
+            dropout_prob,
+            training,
+            fused,
+            manager=None,
+        ):
+            seen_bda_managers.append(manager)
+            return original_residual
+
+        monkeypatch.setattr(layer.hyper_connection, "forward", fake_hyper_connection_forward)
+        monkeypatch.setattr(
+            layer.hyper_connection, "fused_h_res_h_post_bda", fake_fused_h_res_h_post_bda
+        )
+
+        output, _ = layer(hidden_states, attention_mask=None, mhc_recompute_manager=manager)
+        assert output is hidden_states
+        assert seen_bda_managers == [None]
+
+        manager.is_last_layer_in_recompute_block = False
+        layer(hidden_states, attention_mask=None, mhc_recompute_manager=manager)
+        assert seen_bda_managers[-1] is manager
+
+    def test_forward_with_hyper_connections(self):
+        model_config = TransformerConfig(
+            num_layers=3,
+            hidden_size=256,
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+        )
+        model = HybridModel(
+            config=model_config,
+            hybrid_stack_spec=hybrid_stack_spec,
+            vocab_size=100,
+            max_sequence_length=4,
+            hybrid_layer_pattern="M*-",
+        )
+        model.cuda()
+
+        sequence_length = model.max_sequence_length
+        micro_batch_size = 2
+        data = list(range(sequence_length))
+        input_ids = torch.tensor(data, dtype=torch.int64).repeat((micro_batch_size, 1)).cuda()
+        position_ids = torch.tensor(data, dtype=torch.int64).repeat((micro_batch_size, 1)).cuda()
+        attention_mask = torch.ones(
+            (micro_batch_size, 1, sequence_length, sequence_length), dtype=bool
+        ).cuda()
+
+        logits = model.forward(
+            input_ids=input_ids, position_ids=position_ids, attention_mask=attention_mask
+        )
+
+        assert logits.shape[0] == micro_batch_size
+        assert logits.shape[1] == sequence_length
+        assert logits.shape[2] == model.vocab_size
+
+    def test_dummy_hybrid_model_with_hyper_connections_forward_backward(self):
+        model_config = TransformerConfig(
+            num_layers=3,
+            hidden_size=32,
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+            mhc_sinkhorn_iterations=3,
+        )
+        model = HybridModel(
+            config=model_config,
+            hybrid_stack_spec=_get_dummy_hybrid_stack_spec(),
+            vocab_size=64,
+            max_sequence_length=8,
+            hybrid_layer_pattern="M*-",
+            parallel_output=False,
+        )
+
+        assert all(isinstance(layer, HyperConnectionHybridLayer) for layer in model.decoder.layers)
+        assert all(
+            isinstance(layer.inner_layer, _DummyHybridLayer) for layer in model.decoder.layers
+        )
+
+        model.cuda()
+        sequence_length = model.max_sequence_length
+        micro_batch_size = 2
+        data = torch.arange(sequence_length, dtype=torch.int64, device='cuda')
+        input_ids = data.repeat((micro_batch_size, 1))
+        position_ids = data.repeat((micro_batch_size, 1))
+
+        logits = model.forward(input_ids=input_ids, position_ids=position_ids, attention_mask=None)
+
+        assert logits.shape == (micro_batch_size, sequence_length, model.vocab_size)
+        assert torch.isfinite(logits).all()
+
+        logits.float().mean().backward()
+
+        for layer in model.decoder.layers:
+            assert layer.inner_layer.seen_hidden_shapes == [
+                (sequence_length, micro_batch_size, model_config.hidden_size)
+            ]
+            assert layer.inner_layer.proj.weight.grad is not None
+            assert layer.hyper_connection.mapping_proj.weight.grad is not None
+            assert torch.isfinite(layer.inner_layer.proj.weight.grad).all()
+            assert torch.isfinite(layer.hyper_connection.mapping_proj.weight.grad).all()
+
     def test_set_input_tensor(self):
         config: TransformerConfig = self.model.config
         sequence_length = self.model.max_sequence_length
diff --git a/tests/unit_tests/ssm/test_hybrid_block.py b/tests/unit_tests/ssm/test_hybrid_block.py
index 08bf7f2bc28..14caa55aa0a 100644
--- a/tests/unit_tests/ssm/test_hybrid_block.py
+++ b/tests/unit_tests/ssm/test_hybrid_block.py
@@ -3,7 +3,7 @@
 import pytest
 import torch
 
-from megatron.core.models.hybrid.hybrid_block import HybridStack
+from megatron.core.models.hybrid.hybrid_block import HybridStack, HyperConnectionHybridLayer
 from megatron.core.models.hybrid.hybrid_layer_allocation import Symbols, validate_segment_layers
 from megatron.core.models.hybrid.hybrid_layer_specs import hybrid_stack_spec
 from megatron.core.process_groups_config import ProcessGroupCollection
@@ -30,8 +30,13 @@ def setup_method(self, method):
     def get_pg_collection(self):
         return ProcessGroupCollection.use_mpu_process_groups(required_pgs=['tp', 'pp', 'cp'])
 
-    def get_mamba_block(self, layer_pattern):
+    def get_mamba_block(self, layer_pattern, enable_hyper_connections=False):
         layer_type_list = validate_segment_layers(layer_pattern)
+        mhc_kwargs = (
+            {"enable_hyper_connections": True, "hidden_dropout": 0.0, "mhc_sinkhorn_iterations": 5}
+            if enable_hyper_connections
+            else {}
+        )
         transformer_config = TransformerConfig(
             hidden_size=256,  # The Mamba layer places several constraints on this
             # Need to specify num_attention_heads and num_layers or TransformerConfig
@@ -39,6 +44,7 @@ def get_mamba_block(self, layer_pattern):
             num_layers=len(layer_type_list),
             num_attention_heads=4,
             use_cpu_initialization=True,
+            **mhc_kwargs,
         )
         modules = hybrid_stack_spec.submodules
         return HybridStack(
@@ -49,8 +55,13 @@ def get_mamba_block(self, layer_pattern):
             pg_collection=self.get_pg_collection(),
         )
 
-    def get_dsa_mamba_block(self, layer_pattern):
+    def get_dsa_mamba_block(self, layer_pattern, enable_hyper_connections=False):
         layer_type_list = validate_segment_layers(layer_pattern)
+        mhc_kwargs = (
+            {"enable_hyper_connections": True, "hidden_dropout": 0.0, "mhc_sinkhorn_iterations": 5}
+            if enable_hyper_connections
+            else {}
+        )
         transformer_config = MLATransformerConfig(
             hidden_size=256,  # The Mamba layer places several constraints on this
             # Need to specify num_attention_heads and num_layers or TransformerConfig
@@ -71,6 +82,7 @@ def get_dsa_mamba_block(self, layer_pattern):
             dsa_indexer_n_heads=8,
             dsa_indexer_head_dim=64,
             dsa_indexer_topk=32,
+            **mhc_kwargs,
         )
         modules = hybrid_stack_spec.submodules
         return HybridStack(
@@ -118,6 +130,161 @@ def test_layer_types(self):
         assert isinstance(layers[2], TransformerLayer)
         assert isinstance(layers[2].mlp, MLP)
 
+    def test_hyper_connection_layer_wrappers(self):
+        """mHC wraps each hybrid layer while preserving the layer type underneath."""
+        layer_pattern = Symbols.MAMBA + Symbols.ATTENTION + Symbols.MLP
+        block = self.get_mamba_block(layer_pattern, enable_hyper_connections=True)
+        layers = block.layers
+        assert all(isinstance(layer, HyperConnectionHybridLayer) for layer in layers)
+        assert isinstance(layers[0].inner_layer, MambaLayer)
+        assert isinstance(layers[1].inner_layer, TransformerLayer)
+        assert isinstance(layers[1].inner_layer.self_attention, SelfAttention)
+        assert isinstance(layers[2].inner_layer, TransformerLayer)
+        assert isinstance(layers[2].inner_layer.mlp, MLP)
+
+    def test_hyper_connection_recompute_plan_for_hybrid_layers(self):
+        """HybridStack creates per-layer mHC recompute managers when requested."""
+        layer_pattern = Symbols.MAMBA + Symbols.ATTENTION + Symbols.MLP
+        layer_type_list = validate_segment_layers(layer_pattern)
+        transformer_config = TransformerConfig(
+            hidden_size=256,
+            num_layers=len(layer_type_list),
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+            mhc_sinkhorn_iterations=5,
+            recompute_granularity="selective",
+            recompute_modules=["core_attn", "mhc"],
+        )
+        block = HybridStack(
+            transformer_config,
+            hybrid_stack_spec.submodules,
+            layer_type_list=layer_type_list,
+            pp_layer_offset=0,
+            pg_collection=self.get_pg_collection(),
+        )
+
+        managers, block_ends = block._build_mhc_recompute_layer_plan(use_mhc_recompute=True)
+        assert len(managers) == len(block.layers)
+        assert all(manager is not None for manager in managers)
+        assert block_ends[-1] is True
+
+    def test_hyper_connection_gpu_forward(self):
+        """mHC-enabled HybridStack expands internally and contracts back at the output."""
+        layer_pattern = Symbols.MAMBA + Symbols.ATTENTION + Symbols.MLP
+        block = self.get_mamba_block(layer_pattern, enable_hyper_connections=True)
+        block.cuda()
+        micro_batch_size = 2
+        sequence_length = 32
+        hidden_states = torch.ones((sequence_length, micro_batch_size, block.config.hidden_size))
+        hidden_states = hidden_states.cuda()
+        attention_mask = torch.ones(
+            (micro_batch_size, 1, sequence_length, sequence_length), dtype=bool
+        )
+        attention_mask = attention_mask.cuda()
+        output = block(hidden_states, attention_mask=attention_mask)
+        assert output.shape[0] == sequence_length
+        assert output.shape[1] == micro_batch_size
+        assert output.shape[2] == block.config.hidden_size
+        assert output.dtype == torch.float32
+
+    def test_hyper_connection_gdn_gpu_forward(self):
+        """mHC runs through GDN, attention, and Mamba hybrid layers."""
+        layer_pattern = Symbols.GDN + Symbols.ATTENTION + Symbols.MAMBA
+        layer_type_list = validate_segment_layers(layer_pattern)
+        transformer_config = TransformerConfig(
+            hidden_size=256,
+            num_layers=len(layer_type_list),
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            activation_func=torch.nn.functional.silu,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+            mhc_sinkhorn_iterations=5,
+        )
+        block = HybridStack(
+            transformer_config,
+            hybrid_stack_spec.submodules,
+            layer_type_list=layer_type_list,
+            pp_layer_offset=0,
+            pg_collection=self.get_pg_collection(),
+        )
+        block.cuda()
+        micro_batch_size = 2
+        sequence_length = 32
+        hidden_states = torch.ones((sequence_length, micro_batch_size, block.config.hidden_size))
+        hidden_states = hidden_states.cuda()
+        attention_mask = torch.ones(
+            (micro_batch_size, 1, sequence_length, sequence_length), dtype=bool
+        ).cuda()
+        output = block(hidden_states, attention_mask=attention_mask)
+        assert output.shape == (sequence_length, micro_batch_size, block.config.hidden_size)
+
+    def test_hyper_connection_dsa_layer_wrappers(self):
+        """mHC wraps DeepSeek-style DSA and MLP split layers."""
+        layer_pattern = Symbols.MAMBA + Symbols.DS_ATTENTION + Symbols.MLP
+        block = self.get_dsa_mamba_block(layer_pattern, enable_hyper_connections=True)
+        layers = block.layers
+        assert all(isinstance(layer, HyperConnectionHybridLayer) for layer in layers)
+        assert isinstance(layers[0].inner_layer, MambaLayer)
+        assert isinstance(layers[1].inner_layer, TransformerLayer)
+        assert isinstance(layers[1].inner_layer.self_attention, MLASelfAttention)
+        assert isinstance(layers[1].inner_layer.self_attention.core_attention, DSAttention)
+        assert isinstance(layers[2].inner_layer, TransformerLayer)
+        assert isinstance(layers[2].inner_layer.mlp, MLP)
+
+    def test_hyper_connection_pipeline_boundary_shapes(self):
+        """HybridStack keeps n-stream tensors between PP stages and contracts at the end."""
+        layer_type_list = validate_segment_layers(Symbols.MAMBA)
+        transformer_config = TransformerConfig(
+            hidden_size=256,
+            num_layers=len(layer_type_list),
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            hidden_dropout=0.0,
+            mhc_sinkhorn_iterations=5,
+        )
+        modules = hybrid_stack_spec.submodules
+        first_stage = HybridStack(
+            transformer_config,
+            modules,
+            layer_type_list=layer_type_list,
+            pp_layer_offset=0,
+            post_process=False,
+            pg_collection=self.get_pg_collection(),
+        ).cuda()
+        last_stage = HybridStack(
+            transformer_config,
+            modules,
+            pre_process=False,
+            layer_type_list=layer_type_list,
+            pp_layer_offset=1,
+            post_process=True,
+            pg_collection=self.get_pg_collection(),
+        ).cuda()
+
+        micro_batch_size = 2
+        sequence_length = 32
+        hidden_states = torch.ones(
+            (sequence_length, micro_batch_size, transformer_config.hidden_size), device='cuda'
+        )
+        attention_mask = torch.ones(
+            (micro_batch_size, 1, sequence_length, sequence_length), dtype=bool, device='cuda'
+        )
+
+        pp_hidden = first_stage(hidden_states, attention_mask=attention_mask)
+        assert pp_hidden.shape == (
+            sequence_length,
+            micro_batch_size,
+            transformer_config.hidden_size * transformer_config.num_residual_streams,
+        )
+
+        last_stage.set_input_tensor(pp_hidden.detach())
+        output = last_stage(hidden_states, attention_mask=attention_mask)
+        assert output.shape == (sequence_length, micro_batch_size, transformer_config.hidden_size)
+
     def test_invalid_layer_types_cause_failure(self):
         invalid_symbol = '+'
         assert invalid_symbol not in Symbols.VALID_LAYERS  # sanity check.