@@ -705,6 +705,11 @@ def init_state_fn(opt, config=None):
     # TODO(M4): plumb TP groups through optimizer constructors so these setattrs disappear.
     setattr(optimizer, 'tp_group', tp_group)
     setattr(optimizer, 'expert_tp_group', expert_tp_group)
+    # The GTP axes this optimizer's params are sharded over. Taken from the collection rather
+    # than MPU because a MIMO module owns its axes through its own grid and leaves the MPU
+    # globals unset, where the duplicate filter would read rank 0 everywhere.
+    setattr(optimizer, 'gtp_group', getattr(pg_collection, 'gtp_remat', None))
+    setattr(optimizer, 'expert_gtp_group', getattr(pg_collection, 'expt_gtp_remat', None))
 
     return optimizer
 
@@ -930,6 +935,8 @@ def _get_megatron_emerging_optimizer(
             expert_tp_group = getattr(pg_collection, 'expt_tp', tp_group)
             setattr(optimizer, 'tp_group', tp_group)
             setattr(optimizer, 'expert_tp_group', expert_tp_group)
+            setattr(optimizer, 'gtp_group', getattr(pg_collection, 'gtp_remat', None))
+            setattr(optimizer, 'expert_gtp_group', getattr(pg_collection, 'expt_gtp_remat', None))
             results.append(optimizer)
             continue
         else:
 
@@ -226,6 +226,8 @@ def count_zeros_fp32(
     use_decoupled_grad: bool = False,
     tp_group: Optional[torch.distributed.ProcessGroup] = None,
     expert_tp_group: Optional[torch.distributed.ProcessGroup] = None,
+    gtp_group: Optional[torch.distributed.ProcessGroup] = None,
+    expert_gtp_group: Optional[torch.distributed.ProcessGroup] = None,
 ) -> float:
     """Counts the number of zero values in the gradients of the given parameters.
 
@@ -274,7 +276,9 @@ def count_zeros_fp32(
         is_not_tp_duplicate = param_is_not_tensor_parallel_duplicate(
             param, tp_group=tp_group, expert_tp_group=expert_tp_group
         )
-        is_not_gtp_duplicate = param_is_not_gtp_duplicate(param)
+        is_not_gtp_duplicate = param_is_not_gtp_duplicate(
+            param, gtp_group=gtp_group, expert_gtp_group=expert_gtp_group
+        )
         if grad_not_none and is_not_shared and is_not_tp_duplicate and is_not_gtp_duplicate:
             grad_obj = getattr(param, grad_attr)
             data_parallel_group = get_data_parallel_group_if_dtensor(grad_obj, data_parallel_group)
 
@@ -668,11 +668,17 @@ def __init__(
 
         self.tp_group = self.pg_collection.tp
         self.expert_tp_group = getattr(self.pg_collection, 'expt_tp', self.tp_group)
+        # The module's own GTP axes: a MIMO module's grid carries these, and MPU has no
+        # globals to fall back on there (see param_is_not_gtp_duplicate).
+        self.gtp_group = getattr(self.pg_collection, 'gtp_remat', None)
+        self.expert_gtp_group = getattr(self.pg_collection, 'expt_gtp_remat', None)
         for optimizer in optimizers:
             # Child optimizers perform duplicate filtering and gradient-stat reductions.
             optimizer.grad_stats_parallel_group = self.grad_stats_parallel_group
             optimizer.tp_group = self.tp_group
             optimizer.expert_tp_group = self.expert_tp_group
+            optimizer.gtp_group = self.gtp_group
+            optimizer.expert_gtp_group = self.expert_gtp_group
 
         super().__init__(optimizers)
 
@@ -1118,6 +1124,8 @@ def count_zeros(self):
             use_decoupled_grad=self.config.use_precision_aware_optimizer_no_fp8_or_ds_fp8,
             tp_group=self.tp_group,
             expert_tp_group=self.expert_tp_group,
+            gtp_group=self.gtp_group,
+            expert_gtp_group=self.expert_gtp_group,
         )
 
     def start_param_sync_for_bucket_group_subset(self) -> None:
 
@@ -301,7 +301,11 @@ def _filter_grads_for_norm(
                 tp_group=getattr(self, 'tp_group', None),
                 expert_tp_group=getattr(self, 'expert_tp_group', None),
             )
-            is_not_gtp_duplicate = tensor_parallel.param_is_not_gtp_duplicate(param)
+            is_not_gtp_duplicate = tensor_parallel.param_is_not_gtp_duplicate(
+                param,
+                gtp_group=getattr(self, 'gtp_group', None),
+                expert_gtp_group=getattr(self, 'expert_gtp_group', None),
+            )
             if grad_not_none and is_not_shared and is_not_tp_duplicate and is_not_gtp_duplicate:
                 grads_for_norm.append(grad)
         return grads_for_norm
@@ -464,6 +468,8 @@ def count_zeros(self) -> float:
             use_decoupled_grad=self._uses_decoupled_grad(params),
             tp_group=getattr(self, 'tp_group', None),
             expert_tp_group=getattr(self, 'expert_tp_group', None),
+            gtp_group=getattr(self, 'gtp_group', None),
+            expert_gtp_group=getattr(self, 'expert_gtp_group', None),
         )
 
     @abstractmethod
@@ -1901,6 +1907,8 @@ def count_zeros(self):
                 ),
                 tp_group=getattr(self.chained_optimizers[0], 'tp_group', None),
                 expert_tp_group=getattr(self.chained_optimizers[0], 'expert_tp_group', None),
+                gtp_group=getattr(self.chained_optimizers[0], 'gtp_group', None),
+                expert_gtp_group=getattr(self.chained_optimizers[0], 'expert_gtp_group', None),
             )
         else:
             num_zeros_in_grad = 0
 
@@ -121,16 +121,26 @@ def copy_gtp_attributes(destination, source):
             setattr(destination, attr, getattr(source, attr))
 
 
-def param_is_not_gtp_duplicate(param):
+def param_is_not_gtp_duplicate(param, gtp_group=None, expert_gtp_group=None):
     """True if the param's grad is counted once across the GTP_remat/EGTP_remat axis.
 
     GTP_remat/EGTP_remat shards are unique per peer (kept); replicated params counted only on
     rank 0 of the gtp_remat/egtp_remat axis (else counted N times). When GTP_remat is off rank is 0,
     so every param is kept.
+
+    Pass the group that owns the param's GTP axis. The MPU fallback below is only correct for
+    callers whose axis is the global one: a module carrying its own grid (MIMO builds every
+    module's ``gtp_remat`` group in its HyperCommGrid and never initializes the MPU globals)
+    reads rank 0 on every rank there, which keeps every replicated param on every peer.
     """
     if getattr(param, "is_gtp_weight_remat", False):
         return True
     is_expert = not getattr(param, "allreduce", True)
+    group = expert_gtp_group if is_expert else gtp_group
+    # Prefer provided group when available (new explicit path).
+    if group is not None:
+        return group.rank() == 0
+    # Fallback to legacy global state (back-compat).
     if is_expert:
         return get_expert_gtp_weight_remat_rank() == 0
     return get_gtp_weight_remat_rank() == 0
 
@@ -251,7 +251,10 @@ class KernelEntry:
     KernelEntry(
         name="tensor_parallel_layers",
         sources=("megatron/core/tensor_parallel/layers.py",),
-        tests=(K + "test_tensor_parallel_kernels.py",),
+        tests=(
+            K + "test_tensor_parallel_kernels.py",
+            "tests/unit_tests/optimizer/test_grad_norm_gtp_invariance.py",
+        ),
         kind="torch-op",
         notes="VocabParallelEmbedding (weight[idx] deterministic branch vs F.embedding) and local Column/RowParallelLinear "
         "incl. apex fused_weight_gradient_mlp_cuda gradient-accumulation fusion.",
@@ -468,7 +471,10 @@ class KernelEntry:
             "megatron/core/optimizer/optimizer.py",
             "megatron/training/utils/common_utils.py",
         ),
-        tests=(K + "test_optimizer_kernels.py",),
+        tests=(
+            K + "test_optimizer_kernels.py",
+            "tests/unit_tests/optimizer/test_grad_norm_gtp_invariance.py",
+        ),
         kind="external-lib",
         notes="multi_tensor l2norm / scale (TE, apex or local fallback) and fused Adam; "
         "optimizer.py (gradient unscaling) and training/utils/common_utils.py (param / grad norm "