diff --git a/megatron/core/transformer/hyper_connection.py b/megatron/core/transformer/hyper_connection.py
new file mode 100644
index 00000000000..b38811c04fb
--- /dev/null
+++ b/megatron/core/transformer/hyper_connection.py
@@ -0,0 +1,813 @@
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
+import math
+from typing import TYPE_CHECKING, Optional, Tuple
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from torch import Tensor
+
+from megatron.core.transformer.module import MegatronModule
+from megatron.core.transformer.transformer_config import TransformerConfig
+from megatron.core.utils import nvtx_decorator
+
+if TYPE_CHECKING:
+    from megatron.core.tensor_parallel.random import CheckpointManager
+
+_MHC_SINKHORN_EPS = 1e-6
+_MHC_COMPUTE_H_EPS = 1e-6
+
+
+@torch.compile
+def _sinkhorn_iterations(input_logits: Tensor, num_iterations: int, eps: float) -> Tensor:
+    M = input_logits.softmax(dim=-1) + eps
+    M = M / (M.sum(dim=-2, keepdim=True) + eps)
+    for _ in range(num_iterations - 1):
+        M = M / (M.sum(dim=-1, keepdim=True) + eps)
+        M = M / (M.sum(dim=-2, keepdim=True) + eps)
+    return M
+
+
+class SinkhornKnopp(torch.autograd.Function):
+    """Sinkhorn-Knopp projection to doubly stochastic matrix.
+
+    This is an autograd.Function because the iterative forward is re-executed
+    during backward (under torch.enable_grad) so that PyTorch's autograd can
+    differentiate through it without storing all intermediate iteration states.
+    """
+
+    @staticmethod
+    def forward(ctx, input_logits: Tensor, num_iterations: int, eps: float = 1e-6) -> Tensor:
+        """Run Sinkhorn iterations and save inputs for backward recomputation."""
+        M = _sinkhorn_iterations(input_logits, num_iterations, eps)
+        ctx.save_for_backward(input_logits)
+        ctx.num_iterations = num_iterations
+        ctx.eps = eps
+        return M
+
+    @staticmethod
+    def backward(ctx, grad_output: Tensor):
+        """Recompute forward under enable_grad and back-propagate."""
+        (input_logits,) = ctx.saved_tensors
+        with torch.enable_grad():
+            logits = input_logits.detach().requires_grad_(True)
+            M = _sinkhorn_iterations(logits, ctx.num_iterations, ctx.eps)
+            M.backward(grad_output)
+        return logits.grad, None, None
+
+
+def native_sinkhorn(input_logits: Tensor, num_iterations: int, eps: float = 1e-6) -> Tensor:
+    """Native Sinkhorn-Knopp (autograd.Function wrapper)."""
+    return SinkhornKnopp.apply(input_logits, num_iterations, eps)
+
+
+@torch.compile
+def native_h_aggregate(x: Tensor, h_pre: Tensor) -> Tensor:
+    """Native n-stream weighted aggregation: out = sum_j(h_pre_j * x_j)."""
+    return (x * h_pre.unsqueeze(-1)).sum(dim=2)
+
+
+@torch.compile
+def native_h_post_bda(
+    h_res: Tensor, original_residual: Tensor, h_post: Tensor, x: Tensor, bias: Optional[Tensor]
+) -> Tensor:
+    """Native H_res.T @ residual + H_post * (x [+ bias])."""
+    s, b, n, C = original_residual.shape
+    h_res_batched = h_res.view(s * b, n, n)
+    residual_batched = original_residual.view(s * b, n, C)
+    mixed = torch.bmm(h_res_batched.transpose(1, 2), residual_batched).view(s, b, n, C)
+    x_expanded = h_post.unsqueeze(-1) * x.unsqueeze(2)
+    if bias is not None:
+        bias_expanded = h_post.unsqueeze(-1) * bias.view(1, 1, 1, C)
+        return x_expanded + bias_expanded + mixed
+    return x_expanded + mixed
+
+
+@torch.compile
+def native_proj_rms(x: Tensor, weight: Tensor, eps: float = 1e-6) -> Tuple[Tensor, Tensor]:
+    """Native fused projection + RMS normalization."""
+    proj = torch.matmul(x, weight.t())
+    norm = x.norm(dim=-1, keepdim=True)
+    K = x.shape[-1]
+    v = norm / math.sqrt(K) + eps
+    r = 1.0 / v
+    return proj, r
+
+
+@torch.compile
+def native_fused_add_3(a: Tensor, b: Tensor, c: Tensor) -> Tensor:
+    """Native 3-way elementwise add (torch.compile fuses into single kernel)."""
+    return a + b + c
+
+
+class BroadcastTensorFused(torch.autograd.Function):
+    """Split one tensor into 3 autograd-graph children sharing the same storage.
+
+    During backward the three incoming gradients are summed with a caller-
+    supplied fused-add function (cuTile or torch.compile fallback) instead of
+    PyTorch's default sequential accumulation.
+    """
+
+    @staticmethod
+    def forward(ctx, x, fused_add_3_fn):
+        """Return three view aliases and save the fused gradient combiner."""
+        ctx.fused_add_3_fn = fused_add_3_fn
+        return x.view_as(x), x.view_as(x), x.view_as(x)
+
+    @staticmethod
+    def backward(ctx, grad1, grad2, grad3):
+        """Combine gradients from the three broadcast aliases."""
+        grads = [g for g in (grad1, grad2, grad3) if g is not None]
+        if len(grads) == 0:
+            return None, None
+        if len(grads) == 1:
+            return grads[0], None
+        if len(grads) == 2:
+            return grads[0] + grads[1], None
+        return ctx.fused_add_3_fn(grad1, grad2, grad3), None
+
+
+@torch.compile
+def learned_output_contract(
+    hidden_states: Tensor, head_fn: Tensor, base: Tensor, scale: Tensor, n: int, eps: float
+) -> Tensor:
+    """Learned output contraction: n-stream → 1-stream via sigmoid-gated weighted sum."""
+    dtype = hidden_states.dtype
+    hidden_states = hidden_states.to(torch.float32)
+    head_fn = head_fn.to(torch.float32)
+    base = base.to(torch.float32)
+    scale = scale.to(torch.float32)
+    rsqrt = torch.rsqrt(hidden_states.square().mean(-1, keepdim=True) + eps)
+    mixes = F.linear(hidden_states, head_fn) * rsqrt
+    pre = torch.sigmoid(mixes * scale + base) + eps
+    y = torch.sum(pre.unsqueeze(-1) * hidden_states.view(*hidden_states.shape[:-1], n, -1), dim=-2)
+    return y.to(dtype)
+
+
+# ============================================================================
+# HyperConnectionModule
+# ============================================================================
+
+
+# TODO: keep hyper connection in fp32 computation
+class HyperConnectionModule(MegatronModule):
+    """
+    Unified mHC (Manifold-Constrained Hyper-Connections) module.
+
+    Implements the complete mHC propagation:
+        x_{l+1} = H_res^T @ x_l + H_post^T @ F(H_pre @ x_l)
+
+    This module handles:
+    1. Computing learnable mappings: H_pre, H_post, H_res (with Sinkhorn-Knopp projection)
+    2. Aggregation: n-stream → 1-stream (H_pre @ x)
+    3. Expansion: 1-stream → n-stream (H_post^T @ output)
+    4. Residual merge: H_res^T @ x + expanded_output
+    5. Block-level expand/contract for TransformerBlock boundaries
+
+    Args:
+        config: TransformerConfig with hyper-connection fields
+        layer_number: Current layer index for initialization
+    """
+
+    def __init__(self, config: TransformerConfig, layer_number: int):
+        super().__init__(config)
+        self.config = config
+        self.layer_number = layer_number
+        self.n = config.num_residual_streams
+        self.hidden_size = config.hidden_size
+        self.sinkhorn_iterations = config.mhc_sinkhorn_iterations
+        self.sinkhorn_eps = _MHC_SINKHORN_EPS
+        self.compute_h_eps = _MHC_COMPUTE_H_EPS
+
+        # Projection weights for dynamic mappings
+        # Input: [s, b, n*C] -> Output: n^2 + 2n values per token
+        # - H_pre: n values
+        # - H_post: n values
+        # - H_res: n^2 values (before Sinkhorn projection)
+        self.mapping_proj = nn.Linear(
+            self.n * self.hidden_size, self.n * self.n + 2 * self.n, bias=False
+        )
+
+        init_alpha = config.mhc_init_gating_factor
+        # Learnable scaling factors (Eq. 5 in paper)
+        self.alpha_pre = nn.Parameter(torch.full((1,), init_alpha))
+        self.alpha_post = nn.Parameter(torch.full((1,), init_alpha))
+        self.alpha_res = nn.Parameter(torch.full((1,), init_alpha))
+
+        # Static bias terms
+        self.bias = nn.Parameter(torch.zeros(self.n * self.n + 2 * self.n))
+        self.norm_eps = 1e-6
+
+        # Native reference operations. Optimized kernels are added by a separate
+        # optional follow-up without changing this module's public contract.
+        self._fused_add_3_op = native_fused_add_3
+        self._sinkhorn_op = native_sinkhorn
+        self._h_aggregate_op = native_h_aggregate
+        self._h_post_bda_op = native_h_post_bda
+        self._proj_rms_op = native_proj_rms
+        self._proj_rms_compute_h_op = None
+
+        self._init_weights()
+
+    def _init_weights(self) -> None:
+        """Initialize weights for stable training."""
+        nn.init.xavier_uniform_(self.mapping_proj.weight)
+
+        # Set sequence_parallel attribute on parameters for gradient synchronization
+        # across TP ranks when sequence_parallel is enabled.
+        # This is required because HyperConnectionModule uses non-TP-aware layers
+        # (nn.Linear, nn.RMSNorm) whose gradients need to be all-reduced.
+        if self.config.sequence_parallel:
+            setattr(self.mapping_proj.weight, 'sequence_parallel', True)
+            setattr(self.alpha_pre, 'sequence_parallel', True)
+            setattr(self.alpha_post, 'sequence_parallel', True)
+            setattr(self.alpha_res, 'sequence_parallel', True)
+            setattr(self.bias, 'sequence_parallel', True)
+
+    def _projection_and_get_norm(self, x: Tensor) -> Tuple[Tensor, Tensor]:
+        """
+        Projection + RMS normalization.
+
+        Args:
+            x: [s, b, n*C] - n-stream hidden states
+        """
+        s, b, nC = x.shape
+        # The mHC mapping computation runs in FP32: the parameters are kept in
+        # FP32 and the activations are upcast here, then compute_mappings casts
+        # the bounded mixing weights back to the activation dtype.
+        x_2d = x.reshape(s * b, nC).to(torch.float32)
+        weight = self.mapping_proj.weight.to(torch.float32)
+        proj, r = self._proj_rms_op(x_2d, weight, self.norm_eps)
+        return proj.view(s, b, -1), r.view(s, b, 1)
+
+    @torch.compile
+    def _compute_h(self, proj: Tensor, r: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
+        """
+        Compute h from projected hidden states and scaling factors.
+
+        Args:
+            proj: [s, b, n^2 + 2n] - projected hidden states
+            r: [s, b, 1] - scaling factors
+
+        Returns:
+            h_pre: [s, b, n] - aggregation weights
+            h_post: [s, b, n] - expansion weights
+            h_res: [s, b, n^2] - residual mixing logits
+        """
+        alpha_ = torch.cat(
+            [
+                self.alpha_pre.expand(self.n),
+                self.alpha_post.expand(self.n),
+                self.alpha_res.expand(self.n * self.n),
+            ],
+            dim=-1,
+        )
+
+        h = r * proj * alpha_ + self.bias
+        # H_pre = σ(α_pre * (θ_pre @ x̃) + b_pre)
+        h_pre = h[..., : self.n].sigmoid() + self.compute_h_eps  # [s, b, n]
+
+        # H_post = 2σ(α_post * (θ_post @ x̃) + b_post)
+        h_post = h[..., self.n : 2 * self.n].sigmoid() * 2
+        h_res = h[..., 2 * self.n :]
+        return h_pre, h_post, h_res
+
+    @nvtx_decorator(message="HyperConnection::compute_mappings")
+    def compute_mappings(self, x: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
+        """
+        Compute mHC mappings from input hidden states.
+
+        Reference: Eq. (5) and (8) in mHC paper
+
+        Args:
+            x: [s, b, n*C] - n-stream hidden states
+
+        Returns:
+            h_pre: [s, b, n] - aggregation weights (sigmoid activated)
+            h_post: [s, b, n] - expansion weights (2*sigmoid activated)
+            h_res: [s, b, n, n] - residual mixing matrix (doubly stochastic)
+        """
+        s, b, _ = x.shape
+
+        if self._proj_rms_compute_h_op is not None:
+            # Fused path: proj_rms + compute_h in one kernel launch sequence
+            x_2d = x.reshape(s * b, self.n * self.hidden_size)
+            with torch.cuda.nvtx.range("HyperConnection::fused_proj_rms_compute_h"):
+                h_pre, h_post, h_res, _ = self._proj_rms_compute_h_op(
+                    x_2d,
+                    self.mapping_proj.weight,
+                    self.alpha_pre,
+                    self.alpha_post,
+                    self.alpha_res,
+                    self.bias,
+                    self.n,
+                    self.norm_eps,
+                    self.compute_h_eps,
+                )
+            h_pre = h_pre.view(s, b, self.n)
+            h_post = h_post.view(s, b, self.n)
+            h_res = h_res.view(s, b, self.n, self.n)
+        else:
+            # Native path: separate proj_rms + _compute_h
+            with torch.cuda.nvtx.range("HyperConnection::projection_and_get_norm"):
+                proj, r = self._projection_and_get_norm(x)
+            with torch.cuda.nvtx.range("HyperConnection::compute_h"):
+                h_pre, h_post, h_res = self._compute_h(proj, r)
+            h_res = h_res.view(s, b, self.n, self.n)
+
+        h_res = self._sinkhorn_op(
+            h_res, self.sinkhorn_iterations, self.sinkhorn_eps
+        )  # [s, b, n, n]
+
+        # The mixing weights are bounded (sigmoid outputs / doubly stochastic
+        # matrix), so after the FP32 computation they are safe to apply to the
+        # streams in the activation dtype.
+        dtype = x.dtype
+        return h_pre.to(dtype), h_post.to(dtype), h_res.to(dtype)
+
+    @torch.compile
+    def _apply_h_post(self, x: Tensor, h_post: Tensor) -> Tensor:
+        """
+        Core implementation of H_post application to a single tensor.
+
+        Computes: H_post^T @ x
+
+        Args:
+            x: Input tensor, can be either:
+               - [s, b, C] - standard hidden states
+               - [C] - bias tensor (will be broadcast)
+            h_post: [s, b, n] - expansion weights
+
+        Returns:
+            output: [s, b, n*C] - expanded tensor
+        """
+        n = self.n
+        s, b, _ = h_post.shape
+
+        if x.dim() == 1:
+            # x is bias with shape [C], need to broadcast to [s, b, 1, C]
+            C = x.shape[0]
+            x_expanded = x.unsqueeze(0).unsqueeze(0).unsqueeze(0).expand(s, b, 1, C)
+        else:
+            # x is [s, b, C]
+            C = x.shape[-1]
+            x_expanded = x.unsqueeze(2)  # [s, b, 1, C]
+
+        # h_post^T @ x : [s, b, n, 1] * [s, b, 1, C] -> [s, b, n, C]
+        # Using broadcast multiply instead of einsum
+        result = h_post.unsqueeze(-1) * x_expanded
+        return result.view(s, b, n * C)
+
+    @nvtx_decorator(message="HyperConnection::apply_h_post")
+    def apply_h_post(
+        self,
+        x_with_bias: Tuple[Tensor, Optional[Tensor]],
+        h_post: Tensor,
+        manager: Optional['CheckpointManager'] = None,
+    ) -> Tuple[Tensor, Optional[Tensor]]:
+        """
+        Apply H_post to x and optionally bias, with optional checkpointing.
+
+        This is the unified entry point that handles both normal execution
+        and checkpoint-based execution for memory efficiency.
+
+        Args:
+            x_with_bias: Tuple of (x, bias) where:
+                - x: [s, b, C] - hidden states
+                - bias: [C] or None - optional bias tensor
+            h_post: [s, b, n] - expansion weights
+            manager: Optional CheckpointManager for checkpoint management.
+                When provided, wraps _apply_h_post with CheckpointWithoutOutput.
+
+        Returns:
+            Tuple of (x_out, bias_out) where:
+                - x_out: [s, b, n*C] - expanded hidden states
+                - bias_out: [s, b, n*C] or None - expanded bias if input bias was not None
+        """
+        x, bias = x_with_bias
+
+        if manager is not None:
+            from megatron.core.tensor_parallel.random import CheckpointWithoutOutput
+
+            # Checkpoint _apply_h_post to discard the output
+            x_out = CheckpointWithoutOutput(ckpt_manager=manager).checkpoint(
+                self._apply_h_post, x, h_post
+            )
+
+            # Checkpoint _apply_h_post for bias if not None
+            if bias is not None:
+                bias_out = CheckpointWithoutOutput(ckpt_manager=manager).checkpoint(
+                    self._apply_h_post, bias, h_post
+                )
+            else:
+                bias_out = None
+        else:
+            # Normal execution without checkpoint
+            x_out = self._apply_h_post(x, h_post)
+            bias_out = self._apply_h_post(bias, h_post) if bias is not None else None
+
+        return x_out, bias_out
+
+    def aggregate(self, x: Tensor, h_pre: Tensor) -> Tensor:
+        """
+        Aggregate n-stream to 1-stream.
+
+        Args:
+            x: [s, b, n*C] - n-stream hidden states
+            h_pre: [s, b, n] - aggregation weights
+
+        Returns:
+            aggregated: [s, b, C] - single stream hidden states
+        """
+        s, b, _ = x.shape
+        C = self.hidden_size
+        x_streams = x.view(s, b, self.n, C)
+        return self._h_aggregate_op(x_streams, h_pre)
+
+    @torch.compile
+    def apply_h_res(self, h_res: Tensor, residual: Tensor) -> Tensor:
+        """
+        Apply H_res to residual using H_res weights.
+
+        Computes: H_res.T @ residual
+
+        Args:
+            h_res: [s, b, n, n] - residual mixing matrix
+            residual: [s, b, n*C] - n-stream hidden states
+        """
+        s, b, _ = residual.shape
+        n = self.n
+        C = self.hidden_size
+
+        # Reshape for bmm: [s, b, n, n] -> [s*b, n, n]
+        h_res_batched = h_res.view(s * b, n, n)
+        # [s, b, n*C] -> [s, b, n, C] -> [s*b, n, C]
+        residual_batched = residual.view(s, b, n, C).view(s * b, n, C)
+
+        # Batch matrix multiply: [s*b, n, n].T @ [s*b, n, C] -> [s*b, n, C]
+        mixed = torch.bmm(h_res_batched.transpose(1, 2), residual_batched)
+
+        return mixed.view(s, b, n * C)
+
+    def forward(
+        self, hidden_states: Tensor, mhc_recompute_manager: Optional['CheckpointManager'] = None
+    ) -> Tuple[Tensor, Tensor, Tensor, Tensor]:
+        """
+        Full mHC forward pass.
+
+        Uses BroadcastTensorFused to split hidden_states into 3 autograd-graph
+        children so that gradient accumulation from the 3 consumers
+        (compute_mappings, aggregate, fused_h_res_h_post_bda) is handled by a
+        single fused add instead of PyTorch's default sequential accumulation.
+
+        Args:
+            hidden_states: [s, b, n*C] - n-stream hidden states
+            mhc_recompute_manager: Optional CheckpointManager for checkpoint management.
+                When provided, uses _forward_with_checkpoint for memory-efficient execution.
+
+        Returns:
+            A 4-tuple. This is an intentional breaking change from the older
+            3-tuple API because fused_h_res_h_post_bda consumes the residual
+            branch created by BroadcastTensorFused.
+            aggregated: [s, b, C] - aggregated input for layer computation
+            h_res: [s, b, n, n] - residual mixing matrix (for fused kernel)
+            h_post: [s, b, n] - expansion weights
+            residual: [s, b, n*C] - residual view for fused_h_res_h_post_bda
+        """
+        if mhc_recompute_manager is not None:
+            return self._forward_with_checkpoint(hidden_states, mhc_recompute_manager)
+        else:
+            return self._forward_normal(hidden_states)
+
+    def _forward_normal(self, hidden_states: Tensor) -> Tuple[Tensor, Tensor, Tensor, Tensor]:
+        """
+        Normal forward pass without checkpointing.
+
+        Args:
+            hidden_states: [s, b, n*C] - n-stream hidden states
+
+        Returns:
+            aggregated: [s, b, C] - aggregated input for layer computation
+            h_res: [s, b, n, n] - residual mixing matrix (for fused kernel)
+            h_post: [s, b, n] - expansion weights
+            residual: [s, b, n*C] - residual view for fused_h_res_h_post_bda
+        """
+        # Split into 3 views to avoid extra grad accumulations in backward
+        hs_for_mappings, hs_for_aggregate, hs_for_residual = BroadcastTensorFused.apply(
+            hidden_states, self._fused_add_3_op
+        )
+
+        # Compute mappings
+        h_pre, h_post, h_res = self.compute_mappings(hs_for_mappings)
+
+        # Aggregate for layer input
+        with torch.cuda.nvtx.range("HyperConnection::aggregate"):
+            aggregated = self.aggregate(hs_for_aggregate, h_pre)
+
+        return aggregated, h_res, h_post, hs_for_residual
+
+    def _forward_with_checkpoint(
+        self, hidden_states: Tensor, manager: 'CheckpointManager'
+    ) -> Tuple[Tensor, Tensor, Tensor, Tensor]:
+        """
+        Forward pass with checkpointing for memory efficiency.
+
+        compute_mappings is called directly (not checkpointed) since its outputs
+        (h_pre, h_post, h_res) are needed downstream. Only aggregate is wrapped with
+        CheckpointWithoutOutput and auto-registered to the manager.
+        apply_h_res is deferred to fused_h_res_h_post_bda for kernel fusion.
+
+        Args:
+            hidden_states: [s, b, n*C] - n-stream hidden states
+            manager: CheckpointManager for unified recomputation
+
+        Returns:
+            aggregated: [s, b, C] - aggregated input for layer computation
+            h_res: [s, b, n, n] - residual mixing matrix (for fused kernel)
+            h_post: [s, b, n] - expansion weights
+            residual: [s, b, n*C] - residual view for fused_h_res_h_post_bda
+        """
+        from megatron.core.tensor_parallel.random import CheckpointWithoutOutput
+
+        # Split into 3 views to avoid extra grad accumulations in backward
+        hs_for_mappings, hs_for_aggregate, hs_for_residual = BroadcastTensorFused.apply(
+            hidden_states, self._fused_add_3_op
+        )
+
+        h_pre, h_post, h_res = self.compute_mappings(hs_for_mappings)
+
+        # Checkpoint aggregate - auto-registers to manager
+        aggregated = CheckpointWithoutOutput(ckpt_manager=manager).checkpoint(
+            self.aggregate, hs_for_aggregate, h_pre
+        )
+
+        return aggregated, h_res, h_post, hs_for_residual
+
+    # ==================== Block-level utilities ====================
+
+    @staticmethod
+    def input_expand(x: Tensor, n: int) -> Tensor:
+        """
+        Expand 1-stream to n-stream at TransformerBlock entry.
+
+        Simple replication strategy: each stream initialized as a copy of input.
+
+        Args:
+            x: [s, b, C] - single stream hidden states
+            n: Number of residual streams
+
+        Returns:
+            expanded: [s, b, n*C] - n-stream hidden states
+        """
+        s, b, C = x.shape
+        # Replicate input to n streams
+        expanded = x.unsqueeze(2).expand(s, b, n, C).contiguous()
+        return expanded.view(s, b, n * C)
+
+    @staticmethod
+    def output_contract(x: Tensor, n: int) -> Tensor:
+        """
+        Contract n-stream to 1-stream at TransformerBlock exit.
+
+        Simple averaging strategy: average all streams.
+
+        Args:
+            x: [s, b, n*C] - n-stream hidden states
+            n: Number of residual streams
+
+        Returns:
+            contracted: [s, b, C] - single stream hidden states
+        """
+        s, b, nC = x.shape
+        C = nC // n
+        # Average all streams
+        x_streams = x.view(s, b, n, C)
+        contracted = x_streams.mean(dim=2)
+        return contracted
+
+    # ==================== Fused kernel placeholder ====================
+
+    @nvtx_decorator(message="HyperConnection::fused_h_res_h_post_bda")
+    def fused_h_res_h_post_bda(
+        self,
+        h_res: Tensor,
+        original_residual: Tensor,
+        h_post: Tensor,
+        layer_output_with_bias: Tuple[Tensor, Optional[Tensor]],
+        dropout_prob: float,
+        training: bool,
+        fused: bool,
+        manager: Optional['CheckpointManager'] = None,
+    ) -> Tensor:
+        """
+        Fused kernel combining apply_h_res, apply_h_post and bias-dropout-add.
+
+        This is a placeholder for future kernel fusion optimization.
+        Currently implements the operations sequentially using native PyTorch.
+
+        The computation flow is:
+            1. mixed = H_res.T @ original_residual (apply_h_res)
+            2. expanded = H_post^T @ layer_output (apply_h_post)
+            3. output = dropout(expanded + bias) + mixed (bias-dropout-add)
+
+        Args:
+            h_res: [s, b, n, n] - residual mixing matrix
+            original_residual: [s, b, n*C] - n-stream hidden states (before H_res applied)
+            h_post: [s, b, n] - expansion weights
+            layer_output_with_bias: Tuple of (x, bias) where:
+                - x: [s, b, C] - layer output (attention or MLP output)
+                - bias: [C] or None - optional bias tensor
+            dropout_prob: Dropout probability
+            training: Whether in training mode
+            fused: Whether to use fused BDA implementation
+            manager: Optional CheckpointManager for checkpoint management.
+                When provided, each operation is wrapped with CheckpointWithoutOutput.
+
+        Returns:
+            output: [s, b, n*C] - final output after all operations
+        """
+        if manager is not None:
+            return self._fused_h_res_h_post_bda_with_checkpoint(
+                h_res,
+                original_residual,
+                h_post,
+                layer_output_with_bias,
+                dropout_prob,
+                training,
+                fused,
+                manager,
+            )
+        else:
+            return self._fused_h_res_h_post_bda_native(
+                h_res,
+                original_residual,
+                h_post,
+                layer_output_with_bias,
+                dropout_prob,
+                training,
+                fused,
+            )
+
+    def _fused_h_res_h_post_bda_native(
+        self,
+        h_res: Tensor,
+        original_residual: Tensor,
+        h_post: Tensor,
+        layer_output_with_bias: Tuple[Tensor, Optional[Tensor]],
+        dropout_prob: float,
+        training: bool,
+        fused: bool,
+    ) -> Tensor:
+        """
+        h_res, h_post and bda.
+
+        When dropout is zero (or inference), uses a single fused/reference kernel
+        for H_res.T @ residual + H_post * (x + bias). Falls back to unfused
+        implementation when dropout is needed.
+
+        Args:
+            h_res: [s, b, n, n] - residual mixing matrix
+            original_residual: [s, b, n*C] - n-stream hidden states
+            h_post: [s, b, n] - expansion weights
+            layer_output_with_bias: Tuple of (x, bias)
+            dropout_prob: Dropout probability
+            training: Whether in training mode
+            fused: Whether to use fused BDA implementation
+
+        Returns:
+            output: [s, b, n*C] - final output
+        """
+        x, bias = layer_output_with_bias
+
+        if dropout_prob == 0.0 or not training:
+            s, b, _ = original_residual.shape
+            n = self.n
+            C = self.hidden_size
+            orig_reshaped = original_residual.view(s, b, n, C)
+            output = self._h_post_bda_op(h_res, orig_reshaped, h_post, x, bias)
+            return output.view(s, b, n * C)
+
+        from megatron.core.fusions.fused_bias_dropout import get_bias_dropout_add
+
+        with torch.cuda.nvtx.range("HyperConnection::apply_h_res"):
+            mixed = self.apply_h_res(h_res, original_residual)
+        with torch.cuda.nvtx.range("HyperConnection::apply_h_post"):
+            x_expanded = self._apply_h_post(x, h_post)
+            bias_expanded = self._apply_h_post(bias, h_post) if bias is not None else None
+        bda_func = get_bias_dropout_add(training, fused)
+        with torch.cuda.nvtx.range("HyperConnection::bda"):
+            output = bda_func((x_expanded, bias_expanded), mixed, dropout_prob)
+        return output
+
+    @nvtx_decorator(message="HyperConnection::fused_h_res_h_post_bda_with_checkpoint")
+    def _fused_h_res_h_post_bda_with_checkpoint(
+        self,
+        h_res: Tensor,
+        original_residual: Tensor,
+        h_post: Tensor,
+        layer_output_with_bias: Tuple[Tensor, Optional[Tensor]],
+        dropout_prob: float,
+        training: bool,
+        fused: bool,
+        manager: 'CheckpointManager',
+    ) -> Tensor:
+        """
+        Checkpointed variant of _fused_h_res_h_post_bda_native.
+
+        Wraps compute in CheckpointWithoutOutput for activation memory savings.
+        Cannot reuse _native directly because checkpoint requires all args to be
+        positional Tensors; tuple/Optional/scalar args are unpacked or captured
+        via closure instead.
+
+        Args:
+            h_res: [s, b, n, n] - residual mixing matrix
+            original_residual: [s, b, n*C] - n-stream hidden states
+            h_post: [s, b, n] - expansion weights
+            layer_output_with_bias: Tuple of (x, bias)
+            dropout_prob: Dropout probability
+            training: Whether in training mode
+            fused: Whether to use fused BDA implementation
+            manager: CheckpointManager for checkpoint management
+
+        Returns:
+            output: [s, b, n*C] - final output
+        """
+        from megatron.core.tensor_parallel.random import CheckpointWithoutOutput
+
+        x, bias = layer_output_with_bias
+        n = self.n
+        C = self.hidden_size
+
+        # Fast path: no dropout — use fused/reference h_post_bda kernel (same as _native)
+        if dropout_prob == 0.0 or not training:
+
+            def _fused_wrapper(h_res, original_residual, h_post, x, *optional_bias):
+                s, b, _ = original_residual.shape
+                orig_reshaped = original_residual.view(s, b, n, C)
+                b_arg = optional_bias[0] if optional_bias else None
+                return self._h_post_bda_op(h_res, orig_reshaped, h_post, x, b_arg).view(s, b, n * C)
+
+            ckpt = CheckpointWithoutOutput(ckpt_manager=manager)
+            if bias is not None:
+                output = ckpt.checkpoint(_fused_wrapper, h_res, original_residual, h_post, x, bias)
+            else:
+                output = ckpt.checkpoint(_fused_wrapper, h_res, original_residual, h_post, x)
+
+        # Slow path: dropout required — fused kernel does not support dropout,
+        # fall back to sequential apply_h_res + apply_h_post + bda
+        else:
+            from megatron.core.fusions.fused_bias_dropout import get_bias_dropout_add
+
+            bda_func = get_bias_dropout_add(training, fused)
+            has_bias = bias is not None
+
+            def _native_wrapper(h_res, original_residual, h_post, x, *optional_bias):
+                with torch.cuda.nvtx.range("HyperConnection::apply_h_res"):
+                    mixed = self.apply_h_res(h_res, original_residual)
+                with torch.cuda.nvtx.range("HyperConnection::apply_h_post"):
+                    x_expanded = self._apply_h_post(x, h_post)
+                    if has_bias:
+                        bias_expanded = self._apply_h_post(optional_bias[0], h_post)
+                    else:
+                        bias_expanded = None
+                with torch.cuda.nvtx.range("HyperConnection::bda"):
+                    output = bda_func((x_expanded, bias_expanded), mixed, dropout_prob)
+                return output
+
+            ckpt = CheckpointWithoutOutput(ckpt_manager=manager)
+            if has_bias:
+                output = ckpt.checkpoint(_native_wrapper, h_res, original_residual, h_post, x, bias)
+            else:
+                output = ckpt.checkpoint(_native_wrapper, h_res, original_residual, h_post, x)
+
+        return output
+
+
+# ==================== Checkpoint utilities for mHC ====================
+
+
+class HyperConnectionCheckpoint:
+    """
+    Checkpoint utility for mHC intermediate activations.
+
+    Implements the paper's "recomputing strategy" to reduce memory footprint
+    by discarding intermediate n-stream activations and recomputing on-the-fly.
+    """
+
+    @staticmethod
+    def compute_optimal_block_size(num_layers: int, num_streams: int) -> int:
+        """
+        Compute optimal recomputation block size.
+
+        From paper Eq. (20): L_r^* ≈ sqrt(nL/(n+2))
+
+        Args:
+            num_layers: Total number of transformer layers
+            num_streams: Number of residual streams (n)
+
+        Returns:
+            block_size: Optimal block size for checkpointing
+        """
+        block_size = int(math.sqrt(num_streams * num_layers / (num_streams + 2)))
+        return max(1, block_size)
diff --git a/megatron/core/transformer/transformer_config.py b/megatron/core/transformer/transformer_config.py
index 761504e614e..23cd3369b9a 100644
--- a/megatron/core/transformer/transformer_config.py
+++ b/megatron/core/transformer/transformer_config.py
@@ -536,7 +536,7 @@ class TransformerConfig(ModelParallelConfig):
     recompute_modules: Optional[List[str]] = None
     """The submodules to recompute.
     choices: "core_attn", "moe_act", "layernorm", "mla_up_proj", "mlp", "moe",
-    "shared_experts", "gdn_norm_out".
+             "shared_experts", "gdn_norm_out", "mhc".
     default: ["core_attn"].
     "core_attn": recompute the core attention part of the transformer layer.
     "moe_act": recompute the MoE MLP activation function.
@@ -546,7 +546,11 @@ class TransformerConfig(ModelParallelConfig):
     "moe": recompute the MoE layer.
     "shared_experts": recompute the shared experts in the MoE layer.
     "gdn_norm_out": recompute the GatedDeltaNet output norm and HP-to-CP all-to-all.
-    "moe_act", "layernorm", "mla_up_proj", and "gdn_norm_out" use output-discarding checkpointing,
+    "mhc": recompute HyperConnection intermediate activations via
+           CheckpointWithoutOutput + CheckpointManager. Requires
+           enable_hyper_connections=True. Cannot be used with "mlp".
+    "moe_act", "layernorm", "mla_up_proj", "gdn_norm_out", and "mhc" use
+    output-discarding checkpointing,
     "core_attn", "mlp", "moe", and "shared_experts" use normal checkpointing.
     """
 
@@ -1073,6 +1077,35 @@ class TransformerConfig(ModelParallelConfig):
     CudaGraphScope instances deserialized from pre-refactor checkpoints are converted to their
     string names before normalization so existing CUDA_GRAPH_MODULES_DEPRECATIONS handles them."""
 
+    ####################
+    # Hyper-Connection Configuration
+    ####################
+    enable_hyper_connections: bool = False
+    """Enable mHC residual connections."""
+
+    num_residual_streams: int = 4
+    """Number of residual streams (n in paper)."""
+
+    mhc_sinkhorn_iterations: int = 20
+    """Number of Sinkhorn-Knopp iterations for doubly stochastic projection."""
+
+    mhc_init_gating_factor: float = 0.01
+    """Initial value of Gating Factor (alpha in paper)."""
+
+    mhc_recompute_layer_num: Optional[int] = None
+    """Number of layers per MHC recompute block.
+
+    When set, every `mhc_recompute_layer_num` layers form a recompute block. The last layer
+    in each recompute block (i.e., layer_number % mhc_recompute_layer_num == 0 or the final
+    layer in the transformer block) will:
+    - NOT checkpoint its final MLP BDA
+    - Register the unified recompute hook on its MLP BDA output
+    - A new CheckpointManager is created for subsequent layers
+
+    If None, all layers in the transformer block share a single recompute block.
+
+    Must be a positive integer when set."""
+
     ####################
     # miscellaneous
     ####################
@@ -1747,6 +1780,7 @@ def __post_init__(self):
                     "moe",
                     "shared_experts",
                     "gdn_norm_out",
+                    "mhc",
                 }
                 invalid_modules = set(self.recompute_modules) - allowed_modules
                 assert not invalid_modules, (
@@ -1818,6 +1852,55 @@ def __post_init__(self):
             if "moe" not in self.recompute_modules:
                 self.recompute_modules.append("moe")
 
+        # Validation for "mhc" in recompute_modules
+        if self.recompute_granularity == "selective" and "mhc" in self.recompute_modules:
+            if not self.enable_hyper_connections:
+                raise ValueError(
+                    "'mhc' in recompute_modules requires enable_hyper_connections=True."
+                )
+            if "mlp" in self.recompute_modules:
+                raise ValueError(
+                    "'mhc' and 'mlp' in recompute_modules cannot be used together. "
+                    "They use different checkpoint mechanisms that may conflict."
+                )
+            if self.mhc_recompute_layer_num is not None and (
+                isinstance(self.mhc_recompute_layer_num, bool)
+                or not isinstance(self.mhc_recompute_layer_num, int)
+                or self.mhc_recompute_layer_num < 1
+            ):
+                raise ValueError(
+                    "mhc_recompute_layer_num must be a positive integer when "
+                    "'mhc' is in recompute_modules."
+                )
+            if self.fine_grained_activation_offloading and self.offload_modules:
+                # mHC checkpoints wrap input_layernorm (inside attn_norm offload context)
+                # and pre_mlp_layernorm (inside mlp_norm offload context). The unified
+                # recompute hook fires before GroupCommitFunction.backward() initializes
+                # the backward chunk, so tensor_pop hits a None chunk for these modules.
+                # Other offload modules (qkv_linear, core_attn, attn_proj, expert_fc1,
+                # moe_act) live inside self_attention/MLP which are NOT wrapped by mHC
+                # checkpoints, so they are safe to use with mHC recompute.
+                _MHC_CONFLICTING_OFFLOAD_MODULES = {"attn_norm", "mlp_norm"}
+                conflicting = _MHC_CONFLICTING_OFFLOAD_MODULES & set(self.offload_modules)
+                if conflicting:
+                    raise ValueError(
+                        f"'mhc' in recompute_modules is incompatible with "
+                        f"offload_modules {conflicting}. The mHC recompute hook fires "
+                        f"before the offloading backward chunk is initialized for these "
+                        f"modules, causing tensor_pop on a None chunk. Remove "
+                        f"{conflicting} from offload_modules or remove 'mhc' from "
+                        f"recompute_modules."
+                    )
+
+        if self.enable_hyper_connections and not (
+            self.recompute_granularity == "selective" and "mhc" in self.recompute_modules
+        ):
+            warnings.warn(
+                "HyperConnections are enabled but 'mhc' is not in "
+                "recompute_modules with selective recompute. Consider adding 'mhc' to "
+                "recompute_modules with selective recompute to reduce activation memory."
+            )
+
         if self.fine_grained_activation_offloading:
             assert (
                 not self.cpu_offloading
diff --git a/tests/unit_tests/transformer/test_hyper_connection_recompute.py b/tests/unit_tests/transformer/test_hyper_connection_recompute.py
new file mode 100644
index 00000000000..7c6be437b78
--- /dev/null
+++ b/tests/unit_tests/transformer/test_hyper_connection_recompute.py
@@ -0,0 +1,421 @@
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
+"""
+Unit tests for HyperConnection block-level recomputation.
+
+Tests the following functionality:
+1. HyperConnectionModule._forward_with_checkpoint correctness
+2. HyperConnectionModule.apply_h_post with CheckpointManager
+3. Multiple HyperConnectionModules chained with a single CheckpointManager
+4. Partial checkpoint (last layer not checkpointed)
+5. TransformerConfig 'mhc' in recompute_modules option
+"""
+
+import pytest
+import torch
+
+from megatron.core.tensor_parallel.random import CheckpointManager, model_parallel_cuda_manual_seed
+from megatron.core.transformer.hyper_connection import HyperConnectionModule
+from megatron.core.transformer.transformer_config import TransformerConfig
+from tests.unit_tests.test_utilities import Utils
+
+
+class TestHyperConnectionCheckpoint:
+    """Test HyperConnectionModule checkpoint functionality."""
+
+    def setup_method(self, method):
+        Utils.initialize_model_parallel(1, 1)
+        model_parallel_cuda_manual_seed(123)
+
+    def teardown_method(self, method):
+        Utils.destroy_model_parallel()
+
+    def _create_hyper_connection_module(self, hidden_size=64, num_residual_streams=4):
+        """Create a HyperConnectionModule for testing."""
+        config = TransformerConfig(
+            num_layers=2,
+            hidden_size=hidden_size,
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            num_residual_streams=num_residual_streams,
+            mhc_sinkhorn_iterations=5,  # Fewer iterations for faster tests
+            mhc_init_gating_factor=0.01,
+        )
+        module = HyperConnectionModule(config=config, layer_number=1)
+        module.cuda()
+        return module
+
+    def test_apply_h_res_uses_h_res_transpose(self):
+        """apply_h_res should compute H_res.T @ residual."""
+        module = self._create_hyper_connection_module(hidden_size=4, num_residual_streams=2)
+        h_res = torch.tensor([[[[1.0, 2.0], [3.0, 4.0]]]], device='cuda')
+        residual = torch.tensor([[[10.0, 100.0, 3.0, 4.0, 1.0, 2.0, 5.0, 6.0]]], device='cuda')
+        expected = torch.tensor(
+            [[[13.0, 106.0, 18.0, 22.0, 24.0, 208.0, 26.0, 32.0]]], device='cuda'
+        )
+
+        mixed = module.apply_h_res(h_res, residual)
+
+        torch.testing.assert_close(mixed, expected, atol=0.0, rtol=0.0)
+
+    def test_forward_normal_vs_checkpoint_correctness(self):
+        """
+        Test that _forward_with_checkpoint produces the same outputs as _forward_normal.
+        """
+        hidden_size = 64
+        num_streams = 4
+        seq_len = 8
+        batch_size = 2
+
+        module = self._create_hyper_connection_module(hidden_size, num_streams)
+
+        # Create input tensors
+        hidden_states = torch.randn(
+            seq_len, batch_size, num_streams * hidden_size, device='cuda', requires_grad=True
+        )
+        residual = torch.randn(
+            seq_len, batch_size, num_streams * hidden_size, device='cuda', requires_grad=True
+        )
+
+        # Clone inputs for comparison
+        hidden_states_ckpt = hidden_states.detach().clone().requires_grad_(True)
+        residual_ckpt = residual.detach().clone().requires_grad_(True)
+
+        # Forward without checkpoint (reference)
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+        aggregated_ref, h_res_ref, h_post_ref, residual_ref = module._forward_normal(hidden_states)
+        mixed_ref = module.apply_h_res(h_res_ref, residual)
+        loss_ref = aggregated_ref.sum() + mixed_ref.sum() + h_post_ref.sum()
+        loss_ref.backward()
+        grad_hidden_ref = hidden_states.grad.clone()
+        grad_residual_ref = residual.grad.clone()
+
+        # Forward with checkpoint
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+        manager = CheckpointManager()
+        aggregated_ckpt, h_res_ckpt, h_post_ckpt, residual_ckpt_out = (
+            module._forward_with_checkpoint(hidden_states_ckpt, manager)
+        )
+        mixed_ckpt = module.apply_h_res(h_res_ckpt, residual_ckpt)
+        # Calculate loss before discarding outputs
+        loss_ckpt = aggregated_ckpt.sum() + mixed_ckpt.sum() + h_post_ckpt.sum()
+
+        # Register unified recompute hook
+        manager.discard_all_outputs_and_register_unified_recompute(loss_ckpt)
+
+        # Backward pass
+        loss_ckpt.backward()
+        grad_hidden_ckpt = hidden_states_ckpt.grad.clone()
+        grad_residual_ckpt = residual_ckpt.grad.clone()
+
+        # Verify gradients match
+        assert torch.allclose(grad_hidden_ckpt, grad_hidden_ref, atol=1e-5), (
+            f"Hidden states gradients mismatch:\n"
+            f"Checkpoint: {grad_hidden_ckpt}\n"
+            f"Reference: {grad_hidden_ref}"
+        )
+        assert torch.allclose(grad_residual_ckpt, grad_residual_ref, atol=1e-5), (
+            f"Residual gradients mismatch:\n"
+            f"Checkpoint: {grad_residual_ckpt}\n"
+            f"Reference: {grad_residual_ref}"
+        )
+
+    def test_apply_h_post_with_checkpoint(self):
+        """
+        Test that apply_h_post with manager produces correct gradients.
+        """
+        hidden_size = 64
+        num_streams = 4
+        seq_len = 8
+        batch_size = 2
+
+        module = self._create_hyper_connection_module(hidden_size, num_streams)
+
+        # Create input tensors
+        x = torch.randn(seq_len, batch_size, hidden_size, device='cuda', requires_grad=True)
+        bias = torch.randn(hidden_size, device='cuda')
+        h_post = torch.randn(seq_len, batch_size, num_streams, device='cuda', requires_grad=True)
+
+        # Clone inputs
+        x_ckpt = x.detach().clone().requires_grad_(True)
+        h_post_ckpt = h_post.detach().clone().requires_grad_(True)
+
+        # Reference: without checkpoint (manager=None)
+        torch.manual_seed(42)
+        x_out_ref, bias_out_ref = module.apply_h_post((x, bias), h_post, manager=None)
+        loss_ref = x_out_ref.sum()
+        if bias_out_ref is not None:
+            loss_ref = loss_ref + bias_out_ref.sum()
+        loss_ref.backward()
+        grad_x_ref = x.grad.clone()
+        grad_h_post_ref = h_post.grad.clone()
+
+        # With checkpoint (manager provided)
+        torch.manual_seed(42)
+        manager = CheckpointManager()
+        x_out_ckpt, bias_out_ckpt = module.apply_h_post(
+            (x_ckpt, bias), h_post_ckpt, manager=manager
+        )
+        loss_ckpt = x_out_ckpt.sum()
+        if bias_out_ckpt is not None:
+            loss_ckpt = loss_ckpt + bias_out_ckpt.sum()
+
+        manager.discard_all_outputs_and_register_unified_recompute(loss_ckpt)
+        loss_ckpt.backward()
+        grad_x_ckpt = x_ckpt.grad.clone()
+        grad_h_post_ckpt = h_post_ckpt.grad.clone()
+
+        # Verify gradients
+        assert torch.allclose(grad_x_ckpt, grad_x_ref, atol=1e-5)
+        assert torch.allclose(grad_h_post_ckpt, grad_h_post_ref, atol=1e-5)
+
+    def test_forward_with_manager_parameter(self):
+        """
+        Test forward() method with mhc_recompute_manager parameter.
+        """
+        hidden_size = 64
+        num_streams = 4
+        seq_len = 8
+        batch_size = 2
+
+        module = self._create_hyper_connection_module(hidden_size, num_streams)
+
+        # Create input tensors
+        hidden_states = torch.randn(
+            seq_len, batch_size, num_streams * hidden_size, device='cuda', requires_grad=True
+        )
+
+        # Clone inputs
+        hidden_states_ckpt = hidden_states.detach().clone().requires_grad_(True)
+
+        # Reference: forward without manager (uses _forward_normal)
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+        aggregated_ref, h_res_ref, h_post_ref, _ = module.forward(
+            hidden_states, mhc_recompute_manager=None
+        )
+        loss_ref = aggregated_ref.sum() + h_res_ref.sum() + h_post_ref.sum()
+        loss_ref.backward()
+        grad_hidden_ref = hidden_states.grad.clone()
+
+        # With manager (uses _forward_with_checkpoint)
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+        manager = CheckpointManager()
+        aggregated_ckpt, h_res_ckpt, h_post_ckpt, _ = module.forward(
+            hidden_states_ckpt, mhc_recompute_manager=manager
+        )
+        loss_ckpt = aggregated_ckpt.sum() + h_res_ckpt.sum() + h_post_ckpt.sum()
+
+        manager.discard_all_outputs_and_register_unified_recompute(loss_ckpt)
+        loss_ckpt.backward()
+        grad_hidden_ckpt = hidden_states_ckpt.grad.clone()
+
+        # Verify gradients match
+        assert torch.allclose(grad_hidden_ckpt, grad_hidden_ref, atol=1e-5)
+
+
+class TestMHCBlockRecomputeIntegration:
+    """Test CheckpointManager integration with HyperConnection."""
+
+    def setup_method(self, method):
+        Utils.initialize_model_parallel(1, 1)
+        model_parallel_cuda_manual_seed(123)
+
+    def teardown_method(self, method):
+        Utils.destroy_model_parallel()
+
+    def test_multiple_hyper_connections_in_chain(self):
+        """
+        Test that multiple HyperConnectionModules can be chained together
+        with a single CheckpointManager.
+        """
+        hidden_size = 64
+        num_streams = 4
+        seq_len = 8
+        batch_size = 2
+        n_channels = num_streams * hidden_size
+
+        # Create multiple HyperConnection modules (simulating multiple layers)
+        config = TransformerConfig(
+            num_layers=4,
+            hidden_size=hidden_size,
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            num_residual_streams=num_streams,
+            mhc_sinkhorn_iterations=5,
+            mhc_init_gating_factor=0.01,
+        )
+
+        modules = [
+            HyperConnectionModule(config=config, layer_number=i + 1).cuda() for i in range(3)
+        ]
+
+        # Create input tensors
+        hidden_states_ref = torch.randn(
+            seq_len, batch_size, n_channels, device='cuda', requires_grad=True
+        )
+        residual_ref = torch.randn(
+            seq_len, batch_size, n_channels, device='cuda', requires_grad=True
+        )
+
+        hidden_states_ckpt = hidden_states_ref.detach().clone().requires_grad_(True)
+        residual_ckpt = residual_ref.detach().clone().requires_grad_(True)
+
+        # Reference: forward without checkpoint
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+
+        h = hidden_states_ref
+        r = residual_ref
+        for module in modules:
+            agg, h_res, h_post, _ = module.forward(h, mhc_recompute_manager=None)
+            agg, _ = module.apply_h_post((0.1 * agg, None), h_post, manager=None)
+            mixed = module.apply_h_res(h_res, r)  # Apply h_res to get mixed [s, b, n*C]
+            h = agg + mixed
+            r = h
+
+        loss_ref = h.sum()
+        loss_ref.backward()
+        grad_hidden_ref = hidden_states_ref.grad.clone()
+        grad_residual_ref = residual_ref.grad.clone()
+
+        # With checkpoint using single manager
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+
+        manager = CheckpointManager()
+
+        h = hidden_states_ckpt
+        r = residual_ckpt
+        for module in modules:
+            agg, h_res, h_post, _ = module.forward(h, mhc_recompute_manager=manager)
+            agg, _ = module.apply_h_post((0.1 * agg, None), h_post, manager=manager)
+            mixed = module.apply_h_res(h_res, r)  # Apply h_res to get mixed [s, b, n*C]
+            h = agg + mixed
+            r = h
+
+        loss_ckpt = h.sum()
+        manager.discard_all_outputs_and_register_unified_recompute(loss_ckpt)
+        loss_ckpt.backward()
+
+        grad_hidden_ckpt = hidden_states_ckpt.grad.clone()
+        grad_residual_ckpt = residual_ckpt.grad.clone()
+
+        # Verify gradients
+        assert torch.allclose(
+            grad_hidden_ckpt, grad_hidden_ref, atol=1e-4
+        ), f"Chained HyperConnection hidden gradients mismatch"
+        assert torch.allclose(
+            grad_residual_ckpt, grad_residual_ref, atol=1e-4
+        ), f"Chained HyperConnection residual gradients mismatch"
+
+    def test_partial_checkpoint_last_layer_not_checkpointed(self):
+        """
+        Test that when is_last_layer_in_block=True, the final output is NOT checkpointed.
+        This simulates the TransformerBlock behavior where the last layer's MLP BDA
+        serves as the hook_tensor for unified recompute.
+        """
+        hidden_size = 64
+        num_streams = 4
+        seq_len = 8
+        batch_size = 2
+
+        config = TransformerConfig(
+            num_layers=2,
+            hidden_size=hidden_size,
+            num_attention_heads=4,
+            use_cpu_initialization=True,
+            enable_hyper_connections=True,
+            num_residual_streams=num_streams,
+            mhc_sinkhorn_iterations=5,
+            mhc_init_gating_factor=0.01,
+        )
+
+        module = HyperConnectionModule(config=config, layer_number=1).cuda()
+
+        hidden_states_ref = torch.randn(
+            seq_len, batch_size, num_streams * hidden_size, device='cuda', requires_grad=True
+        )
+        residual_ref = torch.randn(
+            seq_len, batch_size, num_streams * hidden_size, device='cuda', requires_grad=True
+        )
+
+        hidden_states_ckpt = hidden_states_ref.detach().clone().requires_grad_(True)
+        residual_ckpt = residual_ref.detach().clone().requires_grad_(True)
+
+        # Reference
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+        aggregated_ref, h_res_ref, h_post_ref, _ = module.forward(
+            hidden_states_ref, mhc_recompute_manager=None
+        )
+        aggregated_ref, _ = module.apply_h_post(
+            (0.1 * aggregated_ref, None), h_post_ref, manager=None
+        )
+        mixed_ref = module.apply_h_res(
+            h_res_ref, residual_ref
+        )  # Apply h_res to get mixed [s, b, n*C]
+        # Simulate BDA that is NOT checkpointed (last layer)
+        output_ref = aggregated_ref + 0.5 * mixed_ref
+        loss_ref = output_ref.sum()
+        loss_ref.backward()
+        grad_hidden_ref = hidden_states_ref.grad.clone()
+
+        # With manager - checkpoint everything except final output
+        torch.manual_seed(42)
+        torch.cuda.manual_seed(42)
+        manager = CheckpointManager()
+        aggregated_ckpt, h_res_ckpt, h_post_ckpt, _ = module.forward(
+            hidden_states_ckpt, mhc_recompute_manager=manager
+        )
+
+        aggregated_ckpt, _ = module.apply_h_post(
+            (0.1 * aggregated_ckpt, None), h_post_ckpt, manager=manager
+        )
+        mixed_ckpt = module.apply_h_res(
+            h_res_ckpt, residual_ckpt
+        )  # Apply h_res to get mixed [s, b, n*C]
+        # Simulate BDA that is NOT checkpointed (last layer) - this is the hook_tensor
+        output_ckpt = aggregated_ckpt + 0.5 * mixed_ckpt
+
+        # Register unified recompute on the output (which is not checkpointed)
+        manager.discard_all_outputs_and_register_unified_recompute(output_ckpt)
+
+        loss_ckpt = output_ckpt.sum()
+        loss_ckpt.backward()
+        grad_hidden_ckpt = hidden_states_ckpt.grad.clone()
+
+        # Verify gradients match
+        assert torch.allclose(grad_hidden_ckpt, grad_hidden_ref, atol=1e-5)
+
+
+class TestTransformerConfigRecomputeMhc:
+    """Test 'mhc' in recompute_modules configuration."""
+
+    def test_config_default_value(self):
+        """Test that 'mhc' is not in recompute_modules by default."""
+        config = TransformerConfig(num_layers=2, hidden_size=64, num_attention_heads=4)
+        assert "mhc" not in config.recompute_modules
+
+    def test_config_enable_mhc_recompute(self):
+        """Test enabling 'mhc' in recompute_modules."""
+        config = TransformerConfig(
+            num_layers=2,
+            hidden_size=64,
+            num_attention_heads=4,
+            enable_hyper_connections=True,
+            num_residual_streams=4,
+            recompute_modules=["core_attn", "mhc"],
+            recompute_granularity='selective',
+        )
+        assert "mhc" in config.recompute_modules
+        assert config.enable_hyper_connections is True
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])