diff --git a/deepspeed/module_inject/auto_tp.py b/deepspeed/module_inject/auto_tp.py
index 277334576037..9448244e98ca 100755
--- a/deepspeed/module_inject/auto_tp.py
+++ b/deepspeed/module_inject/auto_tp.py
@@ -16,6 +16,7 @@
 from .fusedqkv_utils import require_tp_fused_qkvw
 from deepspeed.module_inject.tp_shard import get_shard_size, get_shard_size_list
 from deepspeed.utils import groups
+from deepspeed.utils.logging import print_dist
 from deepspeed.module_inject.layers import is_autotp_training_mode
 from .autotp_config import TPLayerSpec, AutoTPConfig, PartitionType
 
@@ -214,6 +215,8 @@ def __init__(self,
         self.linear_policies = None
         self.conv_linear_layer = False
         self.partition_config = partition_config
+        self._gathered_column_tie_fallbacks_configured = False
+        self._tied_gathered_column_module_names = set()
         TensorParallel_Layer.set_keep_module_on_host(keep_module_on_host)
 
     def in_module_list(module, module_list):
@@ -424,6 +427,17 @@ def _replace_with_config(self, child, name):
         model_type = self._get_model_type()
         spec = self.partition_config.find_matching_spec(param_name, model_type)
 
+        if any(part in ("lm_head", "embed_out") for part in name.split('.')):
+            spec_details = None
+            if spec is not None:
+                spec_details = {
+                    "patterns": spec.patterns,
+                    "partition_type": spec.partition_type.value,
+                    "gather_output": spec.gather_output,
+                }
+            print_dist(
+                f"AutoTP lm_head spec match: parameter={param_name!r}; matched_spec={spec_details!r}", ranks=[0])
+
         if spec is None:
             # No matching spec found
             if self.partition_config.strict_mode:
@@ -461,13 +475,30 @@ def _create_row_parallel_layer(self, module, spec: TPLayerSpec, name: str):
 
     def _create_column_parallel_layer(self, module, spec: TPLayerSpec, name: str):
         """Create column-parallel layer (AllReduce in backward)."""
+        if spec.gather_output and self.mp_size is not None and self.mp_size > 1:
+            output_dim = module.weight.shape[0]
+            if output_dim % self.mp_size != 0:
+                if any(part in ("lm_head", "embed_out") for part in name.split('.')):
+                    print_dist(
+                        f"AutoTP: '{name}' uses gather_output with uneven output dim {output_dim} and tp_size="
+                        f"{self.mp_size}; falling back to legacy LmHeadLinearAllreduce for checkpoint-safe "
+                        "consolidation.",
+                        ranks=[0],
+                    )
+                    return LmHeadLinearAllreduce(module, self.mp_group)
+                raise NotImplementedError(
+                    f"AutoTP gather_output requires output dimension divisible by tp_size. Layer '{name}' has "
+                    f"output dim {output_dim} with tp_size={self.mp_size}.")
+
         if self.conv_linear_layer:
-            return conv_LinearLayer(module, self.mp_group, name=name)
+            return conv_LinearLayer(module, self.mp_group, name=name, gather_output=spec.gather_output)
         # Only use fused-QKV heuristics when no partition_config is provided.
         elif self.partition_config is None and require_tp_fused_qkvw(name, self.mp_size):
             # Check and handle fused qkv for TP
             return fused_LinearLayer(module, self.mp_group, fused_module=self.module)
         if spec.shape is not None:
+            if spec.gather_output:
+                raise NotImplementedError("AutoTP gather_output does not yet support shaped sub-parameter layers.")
             return SubParamLinearLayer(
                 module,
                 self.mp_group,
@@ -475,7 +506,57 @@ def _create_column_parallel_layer(self, module, spec: TPLayerSpec, name: str):
                 partition_dim=spec.get_partition_dim(),
                 name=name,
             )
-        return LinearLayer(module, self.mp_group, name=name)
+        if spec.gather_output:
+            print_dist(f"AutoTP: replacing '{name}' with LinearLayer(gather_output=True)", ranks=[0])
+        return LinearLayer(module, self.mp_group, name=name, gather_output=spec.gather_output)
+
+    def _configure_gathered_column_tie_fallbacks(self):
+        """Configure a replicated fallback for gathered output layers tied to embeddings."""
+        if self._gathered_column_tie_fallbacks_configured or self.partition_config is None:
+            return
+
+        named_modules = list(self.module.named_modules())
+        embeddings = [(name, module) for name, module in named_modules
+                      if isinstance(module, nn.Embedding) and hasattr(module, "weight")]
+        get_input_embeddings = getattr(self.module, "get_input_embeddings", None)
+        if callable(get_input_embeddings):
+            input_embedding = get_input_embeddings()
+            if input_embedding is not None and hasattr(input_embedding, "weight"):
+                input_embedding_name = next(
+                    (name for name, module in named_modules if module is input_embedding),
+                    None,
+                )
+                is_known_embedding = any(module is input_embedding for _, module in embeddings)
+                if input_embedding_name is not None and not is_known_embedding:
+                    embeddings.append((input_embedding_name, input_embedding))
+        if not embeddings:
+            self._gathered_column_tie_fallbacks_configured = True
+            return
+
+        model_type = self._get_model_type()
+        for module_name, module in named_modules:
+            if not module_name or isinstance(module, nn.Embedding) or not hasattr(module, "weight"):
+                continue
+
+            tied_embedding_name = next(
+                (embedding_name for embedding_name, embedding in embeddings if module.weight is embedding.weight),
+                None,
+            )
+            if tied_embedding_name is None:
+                continue
+
+            spec = self.partition_config.find_matching_spec(module_name + ".weight", model_type)
+            if spec is None or spec.partition_type != PartitionType.COLUMN or not spec.gather_output:
+                continue
+
+            self._tied_gathered_column_module_names.update((module_name, tied_embedding_name))
+            print_dist(
+                f"AutoTP: '{module_name}.weight' is tied to '{tied_embedding_name}.weight'; leaving both modules "
+                "replicated because coupled vocabulary-parallel embedding is not supported yet.",
+                ranks=[0],
+            )
+
+        self._gathered_column_tie_fallbacks_configured = True
 
     def _get_model_type(self) -> Optional[str]:
         """Extract model type from module config or class name."""
@@ -570,6 +651,9 @@ def _replace_autoep_shared_experts(self, autoep_layer, autoep_name):
                 self._replace_module(child, full_name, "")
 
     def _replace_module(self, r_module, prev_name='', prev_class_name=''):
+        if prev_name == '' and prev_class_name == '':
+            self._configure_gathered_column_tie_fallbacks()
+
         for name, child in r_module.named_children():
             if getattr(child, "_is_autoep_layer", False):
                 full_name = prev_name + '.' + name if prev_name else name
@@ -595,7 +679,9 @@ def _replace_module(self, r_module, prev_name='', prev_class_name=''):
             # instead of linear_policies. This keeps all pattern logic centralized here.
             if self.partition_config is not None:
                 full_name = class_name + '.' + name if class_name else name
-                if isinstance(child, nn.Embedding):
+                if full_name in self._tied_gathered_column_module_names:
+                    continue
+                elif isinstance(child, nn.Embedding):
                     # Check if embedding matches any pattern
                     param_name = full_name + ".weight"
                     model_type = self._get_model_type()
diff --git a/deepspeed/module_inject/autotp_config.py b/deepspeed/module_inject/autotp_config.py
index 4bafea806829..26d1238d81f0 100644
--- a/deepspeed/module_inject/autotp_config.py
+++ b/deepspeed/module_inject/autotp_config.py
@@ -49,6 +49,13 @@ class TPLayerSpec:
             partition_type=PartitionType.COLUMN,
         )
 
+        # Column-parallel layer with replicated output (e.g., an untied LM head)
+        TPLayerSpec(
+            patterns=[".*\\.lm_head$"],
+            partition_type=PartitionType.COLUMN,
+            gather_output=True,
+        )
+
         # Fused QKV - GLM style [Q, K, V] concatenated on dim 0
         TPLayerSpec(
             patterns=[".*\\.query_key_value\\.weight$"],
@@ -117,6 +124,9 @@ class TPLayerSpec:
     # Optional: model type constraint (only apply for specific models)
     model_types: Optional[List[str]] = None
 
+    # Gather column-parallel output shards so every TP rank receives the full output
+    gather_output: bool = False
+
     def __post_init__(self):
         if isinstance(self.partition_type, str):
             self.partition_type = PartitionType(self.partition_type.lower())
@@ -285,6 +295,7 @@ def from_dict(cls, config_dict: dict) -> "AutoTPConfig":
                 TPLayerSpec(
                     patterns=spec_dict.get("patterns", []),
                     partition_type=partition_type,
+                    gather_output=spec_dict.get("gather_output", False),
                     shape=shape,
                     partition_dim=spec_dict.get("partition_dim"),
                     model_types=spec_dict.get("model_types"),
diff --git a/deepspeed/module_inject/layers.py b/deepspeed/module_inject/layers.py
index 0a8cda1d1daa..33b1fbe3dbd0 100644
--- a/deepspeed/module_inject/layers.py
+++ b/deepspeed/module_inject/layers.py
@@ -228,6 +228,48 @@ def backward(ctx: Any, grad_output: torch.Tensor) -> Tuple[None, torch.Tensor]:
         return None, grad_output
 
 
+class GatherFromTensorParallelRegion(torch.autograd.Function):
+    """Gather last-dimension shards while keeping the output replicated."""
+
+    @staticmethod
+    def forward(ctx: Any, group: dist.ProcessGroup, input: torch.Tensor) -> torch.Tensor:
+        ctx.group = group
+        if group is None:
+            ctx.partition_sizes = (input.shape[-1], )
+            ctx.tp_index = 0
+            return input
+
+        tp_world_size = dist.get_world_size(group=group)
+        ctx.tp_index = dist.get_rank(group=group)
+        if tp_world_size == 1:
+            ctx.partition_sizes = (input.shape[-1], )
+            return input
+
+        local_size = torch.tensor([input.shape[-1]], dtype=torch.long, device=input.device)
+        gathered_sizes = [torch.empty_like(local_size) for _ in range(tp_world_size)]
+        dist.all_gather(gathered_sizes, local_size, group=group)
+        ctx.partition_sizes = tuple(int(size.item()) for size in gathered_sizes)
+
+        max_partition_size = max(ctx.partition_sizes)
+        if input.shape[-1] == max_partition_size:
+            input_padded = input.contiguous()
+        else:
+            padded_shape = (*input.shape[:-1], max_partition_size)
+            input_padded = input.new_zeros(padded_shape)
+            input_padded[..., :input.shape[-1]].copy_(input)
+
+        gathered = [torch.empty_like(input_padded) for _ in range(tp_world_size)]
+        dist.all_gather(gathered, input_padded, group=group)
+        return torch.cat([shard[..., :size] for shard, size in zip(gathered, ctx.partition_sizes)], dim=-1)
+
+    @staticmethod
+    def backward(ctx: Any, grad_output: torch.Tensor) -> Tuple[None, torch.Tensor]:
+        shard_offset = sum(ctx.partition_sizes[:ctx.tp_index])
+        shard_size = ctx.partition_sizes[ctx.tp_index]
+        grad_input = grad_output.narrow(-1, shard_offset, shard_size).contiguous()
+        return None, grad_input
+
+
 class TensorParallel_Layer(nn.Module, ABC):
     """
     A base class for model layers with  tensor parallelism support.
@@ -677,10 +719,11 @@ def _mark_uc_metadata(self):
 #remove kwargs from partition.
 class LinearLayer(TensorParallel_Layer):
 
-    def __init__(self, module, mp_group=None, skip_partition=False, **kwargs):
+    def __init__(self, module, mp_group=None, skip_partition=False, gather_output=False, **kwargs):
         super(LinearLayer, self).__init__(mp_group, **kwargs)
         self.weight = module.weight
         self.bias = module.bias
+        self.gather_output = gather_output
         if not skip_partition and self._should_materialize_tp_partition():
             self._tp_partition([self.weight, self.bias])
         self.support_training = True
@@ -699,6 +742,9 @@ def forward(self, input):
         else:
             output = AsyncColumnParallel.apply(self.mp_group, input, self.weight, self.bias)
 
+        if self.gather_output:
+            output = GatherFromTensorParallelRegion.apply(self.mp_group, output)
+
         return output
 
     @torch.no_grad()
@@ -765,7 +811,7 @@ def _mark_uc_metadata(self):
 
     # for bwc
     @classmethod
-    def from_weights(cls, weight_shape=None, dtype=torch.half, weight=None, bias=None):
+    def from_weights(cls, weight_shape=None, dtype=torch.half, weight=None, bias=None, gather_output=False):
         if weight is not None:
             in_features = weight.shape[1]
             out_features = weight.shape[0]
@@ -777,7 +823,7 @@ def from_weights(cls, weight_shape=None, dtype=torch.half, weight=None, bias=Non
             in_features = weight_shape[1]
             out_features = weight_shape[0]
             linear = nn.Linear(in_features, out_features, bias=(bias is not None))
-        return cls(linear, skip_partition=True)
+        return cls(linear, skip_partition=True, gather_output=gather_output)
 
 
 class FusedModuleWrapper:
diff --git a/deepspeed/module_inject/tp_plan_converter.py b/deepspeed/module_inject/tp_plan_converter.py
index bed0567e4445..6723c4ee8ccd 100644
--- a/deepspeed/module_inject/tp_plan_converter.py
+++ b/deepspeed/module_inject/tp_plan_converter.py
@@ -9,7 +9,8 @@
 
 logger = logging.getLogger(__name__)
 
-SUPPORTED_STYLES = {"colwise", "rowwise"}
+SUPPORTED_STYLES = {"colwise", "colwise_rep", "colwise_gather_output", "rowwise"}
+# `colwise_rep` was renamed to `colwise_gather_output` in huggingface/transformers#42809.
 
 
 class TPPlanConverter:
@@ -33,10 +34,13 @@ def convert(hf_tp_plan: Dict[str, str]) -> Optional[List[TPLayerSpec]]:
 
         for pattern, partition in hf_tp_plan.items():
             regex_pattern = TPPlanConverter._wildcard_to_regex(pattern)
+            partition_style = partition.lower()
+            gather_output = False
 
-            if partition.lower() == "colwise":
+            if partition_style in ("colwise", "colwise_rep", "colwise_gather_output"):
                 partition_type = PartitionType.COLUMN
-            elif partition.lower() == "rowwise":
+                gather_output = partition_style != "colwise"
+            elif partition_style == "rowwise":
                 partition_type = PartitionType.ROW
 
             # Only add .weight suffix if not already present
@@ -48,6 +52,7 @@ def convert(hf_tp_plan: Dict[str, str]) -> Optional[List[TPLayerSpec]]:
             layer_specs.append(TPLayerSpec(
                 patterns=[regex_pattern],
                 partition_type=partition_type,
+                gather_output=gather_output,
             ))
 
         return layer_specs
diff --git a/deepspeed/runtime/engine.py b/deepspeed/runtime/engine.py
index fdb9975e7d7d..4c096771f667 100755
--- a/deepspeed/runtime/engine.py
+++ b/deepspeed/runtime/engine.py
@@ -135,7 +135,7 @@
 from ..git_version_info import version
 
 from deepspeed.profiling.flops_profiler.profiler import FlopsProfiler
-from deepspeed.utils.logging import print_json_dist, print_configuration, set_log_level_from_string
+from deepspeed.utils.logging import print_dist, print_json_dist, print_configuration, set_log_level_from_string
 
 from deepspeed.accelerator import get_accelerator
 
@@ -703,6 +703,43 @@ def _apply_autotp_partitioning(self, model, tp_config):
         if hasattr(tp_config, "get_partition_config_object"):
             partition_config = tp_config.get_partition_config_object()
 
+        model_config = getattr(model, "config", None)
+        base_tp_plan = getattr(model_config, "base_model_tp_plan", None) if model_config is not None else None
+        class_tp_plan = getattr(type(model), "_tp_plan", None)
+        runtime_tp_plan = getattr(model, "__dict__", {}).get("_tp_plan")
+        from deepspeed.runtime.tensor_parallel.config import _get_hf_tp_plan
+        hf_tp_plan = _get_hf_tp_plan(model)
+
+        def lm_head_entries(tp_plan):
+            if not isinstance(tp_plan, dict):
+                return {}
+            return {
+                pattern: style
+                for pattern, style in tp_plan.items()
+                if any(part in ("lm_head", "embed_out") for part in pattern.split('.'))
+            }
+
+        lm_head_modules = [
+            name for name, _ in model.named_modules()
+            if name and any(part in ("lm_head", "embed_out") for part in name.split('.'))
+        ]
+        selected_route = "custom partition_config" if partition_config is not None else "HuggingFace tp_plan or AutoTP"
+        model_class = f"{type(model).__module__}.{type(model).__qualname__}"
+        print_dist(
+            f"AutoTP tp_plan diagnostics: model_class={model_class}; route={selected_route}; "
+            f"base_model_tp_plan={base_tp_plan!r}; type(model)._tp_plan={class_tp_plan!r}; "
+            f"instance_tp_plan={runtime_tp_plan!r}; "
+            f"effective_tp_plan={hf_tp_plan!r}",
+            ranks=[0],
+        )
+        print_dist(
+            f"AutoTP lm_head diagnostics: modules={lm_head_modules!r}; "
+            f"base_entries={lm_head_entries(base_tp_plan)!r}; class_entries={lm_head_entries(class_tp_plan)!r}; "
+            f"runtime_entries={lm_head_entries(runtime_tp_plan)!r}; "
+            f"effective_entries={lm_head_entries(hf_tp_plan)!r}",
+            ranks=[0],
+        )
+
         if partition_config is not None:
             autotp = AutoTP(module=model,
                             all_reduce_linears=(),
@@ -723,19 +760,22 @@ def _apply_autotp_partitioning(self, model, tp_config):
             setattr(model, "ds_autotp_parsed", True)
             return
 
-        model_config = getattr(model, "config", None)
         from deepspeed.module_inject import replace_transformer_layer
-
-        from deepspeed.runtime.tensor_parallel.config import _get_hf_tp_plan
-
-        hf_tp_plan = _get_hf_tp_plan(model)
         if hf_tp_plan:
             from deepspeed.module_inject.tp_plan_converter import TPPlanConverter
             from deepspeed.module_inject.autotp_config import AutoTPConfig
 
             layer_specs = TPPlanConverter.convert(hf_tp_plan)
             if layer_specs is not None:
-                logger.info(f"Using HuggingFace tp_plan with {len(layer_specs)} layer specifications")
+                gathered_output_patterns = [
+                    pattern for pattern, style in hf_tp_plan.items()
+                    if style.lower() in ("colwise_rep", "colwise_gather_output")
+                ]
+                print_dist(
+                    f"Using HuggingFace tp_plan with {len(layer_specs)} layer specifications; "
+                    f"gathered column output patterns={gathered_output_patterns}",
+                    ranks=[0],
+                )
                 tp_plan_config = AutoTPConfig(tp_size=tp_size, layer_specs=layer_specs)
                 autotp = AutoTP(
                     module=model,
@@ -752,6 +792,14 @@ def _apply_autotp_partitioning(self, model, tp_config):
                 autotp._replace_module(model)
                 setattr(model, "ds_autotp_parsed", True)
                 return
+            print_dist(
+                f"AutoTP: effective HuggingFace tp_plan could not be converted; falling back to heuristic AutoTP. "
+                f"styles={sorted(set(hf_tp_plan.values()))!r}",
+                ranks=[0],
+            )
+        else:
+            print_dist("AutoTP: no effective HuggingFace tp_plan was found; falling back to heuristic AutoTP.",
+                       ranks=[0])
 
         parser_dict = AutoTP.tp_parser(model)
         for client_module, injection_policy in parser_dict:
diff --git a/deepspeed/runtime/tensor_parallel/config.py b/deepspeed/runtime/tensor_parallel/config.py
index 56abf5868d5d..fb0ac10b9fcd 100644
--- a/deepspeed/runtime/tensor_parallel/config.py
+++ b/deepspeed/runtime/tensor_parallel/config.py
@@ -149,15 +149,26 @@ def get_tensor_parallel_config(ds_config):
 def _get_hf_tp_plan(model):
     """Extract tp_plan from HuggingFace model.
 
-    Prefer base_model_tp_plan (from model config) over _tp_plan (runtime attribute)
-    because _tp_plan often contains duplicate entries with a 'model.' prefix added
-    by HuggingFace, which causes spurious duplicate-match warnings during conversion.
+    Merge plans from the model config, model class, and runtime instance.
+    HuggingFace may replace the instance plan with an expanded base-model plan,
+    while model-level entries such as lm_head remain only on the model class.
     """
     config = getattr(model, 'config', None)
-    if config and getattr(config, 'base_model_tp_plan', None):
-        return model.config.base_model_tp_plan
-
-    if getattr(model, '_tp_plan', None):
-        return model._tp_plan
-
-    return None
+    base_plan = getattr(config, 'base_model_tp_plan', None) if config else None
+    class_plan = getattr(type(model), '_tp_plan', None)
+    instance_dict = getattr(model, '__dict__', {})
+    runtime_plan = instance_dict.get('_tp_plan')
+
+    merged_plan = {}
+    canonical_patterns = set()
+    for plan in (base_plan, class_plan, runtime_plan):
+        if not isinstance(plan, dict):
+            continue
+        for pattern, style in plan.items():
+            canonical_pattern = pattern[len('model.'):] if pattern.startswith('model.') else pattern
+            if pattern.startswith('model.') and canonical_pattern in canonical_patterns:
+                continue
+            merged_plan[pattern] = style
+            canonical_patterns.add(canonical_pattern)
+
+    return merged_plan or None
diff --git a/docs/_tutorials/autotp-training.md b/docs/_tutorials/autotp-training.md
index 7f8d2cbd52df..9b0b1f74c6a8 100644
--- a/docs/_tutorials/autotp-training.md
+++ b/docs/_tutorials/autotp-training.md
@@ -118,10 +118,22 @@ For models that define a `tp_plan`, you only need a minimal config:
 ```
 
 DeepSpeed will read the model's `tp_plan` at initialization and convert it to
-internal partition rules. Currently `colwise` and `rowwise` partition types
-are supported. Additional types defined by HuggingFace (such as
-`colwise_rep`, `local_colwise`, `local_rowwise`, etc.) are not yet handled
-and will raise an error if encountered.
+internal partition rules. The supported types are `colwise`, `rowwise`,
+and `colwise_gather_output`(`colwise_rep`). The gathered column styles shard
+the linear weight along its output dimension and AllGather the local output
+shards so every tensor-parallel rank receives the complete output.
+
+Gathered column parallelism currently supports untied output layers. If an
+output layer such as `lm_head` shares the same runtime `Parameter` object with
+an embedding, DeepSpeed leaves both modules replicated and applies tensor
+parallelism to the remaining matched layers. This preserves the tie without
+silently cloning the weight, but does not reduce the embedding or output-layer
+memory footprint. A coupled vocabulary-parallel embedding is required to shard
+the tied weight and is not yet implemented. This fallback uses actual `Parameter`
+identity rather than model configuration metadata such as `tie_word_embeddings`.
+
+Additional HuggingFace types such as `local_colwise` and `local_rowwise` are
+not yet handled and fall back to AutoTP preset-based partitioning.
 
 If you need to override the model's built-in `tp_plan`, provide a
 `partition_config` in the DeepSpeed config -- it takes precedence.
diff --git a/tests/unit/model_parallelism/test_autotp_training.py b/tests/unit/model_parallelism/test_autotp_training.py
index dcd90076d48f..f074060b3c83 100644
--- a/tests/unit/model_parallelism/test_autotp_training.py
+++ b/tests/unit/model_parallelism/test_autotp_training.py
@@ -379,7 +379,7 @@ def process_linear_layer(hidden_dim, input):
     return torch_linear, torch_out
 
 
-def run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel, use_tp_model_init=False):
+def run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel, use_tp_model_init=False, gather_output=False):
     skip_on_device()
     hidden_dim = 128
     batch_size_per_device = 1
@@ -406,6 +406,7 @@ def run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel, use_tp_model
         "layer_specs": [{
             "patterns": [".*\\.weight$"],
             "partition_type": partition_type,
+            "gather_output": gather_output,
         }],
     }
     if preferred_dtype() is torch.float16:
@@ -436,12 +437,16 @@ def run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel, use_tp_model
     # rely on the model's AutoTP-partitioned parameters.
     torch_linear, torch_out = process_linear_layer(hidden_dim, input)
     if column_parallel:
-        linear = LinearLayer(deepcopy(torch_linear), groups.get_tensor_model_parallel_group())
+        linear = LinearLayer(deepcopy(torch_linear),
+                             groups.get_tensor_model_parallel_group(),
+                             gather_output=gather_output)
         out = linear(input.to(get_accelerator().current_device()))
         loss = out.sum()
         loss.backward()
 
-        cur_device_out = torch.chunk(torch_out, tp_size, dim=-1)[groups.get_tensor_model_parallel_rank()]
+        expected_out = torch_out
+        if not gather_output:
+            expected_out = torch.chunk(torch_out, tp_size, dim=-1)[groups.get_tensor_model_parallel_rank()]
         torch_grad = torch.chunk(torch_linear.weight.grad, tp_size, dim=0)[groups.get_tensor_model_parallel_rank()]
         torch_bias_grad = torch.chunk(torch_linear.bias.grad, tp_size, dim=0)[groups.get_tensor_model_parallel_rank()]
 
@@ -453,7 +458,7 @@ def run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel, use_tp_model
                                    torch_grad.to(get_accelerator().current_device()),
                                    atol=1e-3,
                                    rtol=1e-3)
-        torch.testing.assert_close(cur_device_out.to(get_accelerator().current_device()).contiguous(),
+        torch.testing.assert_close(expected_out.to(get_accelerator().current_device()).contiguous(),
                                    out.contiguous(),
                                    atol=1e-2,
                                    rtol=1e-2)
@@ -490,6 +495,9 @@ def testRowParallel(self, tp_size: int, tp_overlap_comm: bool):
     def testColumnParallel(self, tp_size: int, tp_overlap_comm: bool):
         run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel=True)
 
+    def testGatheredColumnParallel(self, tp_size: int, tp_overlap_comm: bool):
+        run_tp_layer_fwd_bwd(tp_size, tp_overlap_comm, column_parallel=True, gather_output=True)
+
 
 # @pytest.mark.sequential
 class TestParamsGather(DistributedTest):
diff --git a/tests/unit/model_parallelism/test_tp_plan_real_models.py b/tests/unit/model_parallelism/test_tp_plan_real_models.py
index 7bea8643700e..0268ed4cfb44 100644
--- a/tests/unit/model_parallelism/test_tp_plan_real_models.py
+++ b/tests/unit/model_parallelism/test_tp_plan_real_models.py
@@ -8,6 +8,8 @@
 import deepspeed.comm as dist
 import deepspeed
 from deepspeed.accelerator import get_accelerator
+from deepspeed.module_inject.layers import LinearLayer
+from deepspeed.runtime.tensor_parallel.config import _get_hf_tp_plan
 from deepspeed.utils import groups
 from unit.common import DistributedTest
 
@@ -23,26 +25,30 @@ class TestTPPlanRealHFModels(DistributedTest):
     world_size = 2
 
     def test_qwen2_tp_plan_with_zero2(self):
-        """Test Qwen2 model + tp_plan + ZeRO2"""
+        """Test an untied Qwen2 LM head with gathered column output and ZeRO2."""
         skip_on_device()
 
         try:
-            from transformers import AutoModelForCausalLM, AutoConfig
+            from transformers import AutoModelForCausalLM, Qwen2Config
         except ImportError:
             pytest.skip("transformers not installed")
 
-        # Create small Qwen2 config
-        config = AutoConfig.from_pretrained(
-            "Qwen/Qwen2-7B",
+        # Construct locally so the test does not download model configuration or weights.
+        config = Qwen2Config(
             vocab_size=1000,
             hidden_size=128,
             intermediate_size=256,
             num_hidden_layers=2,
             num_attention_heads=4,
             num_key_value_heads=4,
+            tie_word_embeddings=False,
         )
 
         model = AutoModelForCausalLM.from_config(config)
+        assert model.lm_head.weight is not model.model.embed_tokens.weight
+
+        tp_plan = _get_hf_tp_plan(model)
+        assert tp_plan["lm_head"] in ("colwise_rep", "colwise_gather_output")
 
         ds_config = {
             "train_micro_batch_size_per_gpu": 1,
@@ -67,6 +73,10 @@ def test_qwen2_tp_plan_with_zero2(self):
         engine, _, _, _ = deepspeed.initialize(model=model, model_parameters=model.parameters(), config=ds_config)
 
         assert engine.autotp_size() == 2
+        assert isinstance(model.lm_head, LinearLayer)
+        assert model.lm_head.gather_output
+        assert model.lm_head.weight.shape == (500, 128)
+        assert model.model.embed_tokens.weight.shape == (1000, 128)
 
         # Train for a few steps
         for _ in range(3):
@@ -77,11 +87,61 @@ def test_qwen2_tp_plan_with_zero2(self):
                 group=groups.get_tensor_model_parallel_group(),
             )
             outputs = engine(input_ids, labels=input_ids)
+            assert outputs.logits.shape == (1, 16, 1000)
             engine.backward(outputs.loss)
             engine.step()
 
             assert not torch.isnan(outputs.loss)
 
+    def test_qwen2_tied_lm_head_falls_back_to_replicated(self):
+        """Test that an actual Qwen2 Parameter tie remains replicated."""
+        skip_on_device()
+
+        try:
+            from transformers import AutoModelForCausalLM, Qwen2Config
+        except ImportError:
+            pytest.skip("transformers not installed")
+
+        config = Qwen2Config(
+            vocab_size=1000,
+            hidden_size=128,
+            intermediate_size=256,
+            num_hidden_layers=1,
+            num_attention_heads=4,
+            num_key_value_heads=4,
+            tie_word_embeddings=True,
+        )
+        model = AutoModelForCausalLM.from_config(config)
+        assert model.lm_head.weight is model.model.embed_tokens.weight
+
+        ds_config = {
+            "train_micro_batch_size_per_gpu": 1,
+            "tensor_parallel": {
+                "autotp_size": 2
+            },
+            "zero_optimization": {
+                "stage": 0
+            },
+        }
+
+        engine, _, _, _ = deepspeed.initialize(model=model, model_parameters=model.parameters(), config=ds_config)
+
+        assert engine.autotp_size() == 2
+        assert isinstance(model.model.layers[0].self_attn.q_proj, LinearLayer)
+        assert isinstance(model.model.embed_tokens, torch.nn.Embedding)
+        assert isinstance(model.lm_head, torch.nn.Linear)
+        assert model.lm_head.weight is model.model.embed_tokens.weight
+        assert model.lm_head.weight.shape == (1000, 128)
+
+        input_ids = torch.randint(0, 1000, (1, 16)).to(get_accelerator().current_device_name())
+        dist.broadcast(
+            input_ids,
+            src=groups.get_tensor_model_parallel_src_rank(),
+            group=groups.get_tensor_model_parallel_group(),
+        )
+        outputs = engine(input_ids)
+        assert outputs.logits.shape == (1, 16, 1000)
+
     def test_custom_model_with_custom_tp_plan(self):
         """Test custom model + custom tp_plan"""
         skip_on_device()
diff --git a/tests/unit/module_inject/test_tp_partition_config_path.py b/tests/unit/module_inject/test_tp_partition_config_path.py
index 5a7c32e9ad03..3e952e439b87 100644
--- a/tests/unit/module_inject/test_tp_partition_config_path.py
+++ b/tests/unit/module_inject/test_tp_partition_config_path.py
@@ -12,6 +12,7 @@
 import torch.nn as nn
 
 from deepspeed.module_inject.auto_tp import AutoTP, AutoTPConfig, PartitionType, TPLayerSpec
+from deepspeed.module_inject.layers import LinearLayer, LmHeadLinearAllreduce
 
 
 class SubAttn(nn.Module):
@@ -41,6 +42,17 @@ def __init__(self, num_layers=2):
         self.head = nn.Linear(32, 100, bias=False)
 
 
+class OutputModel(nn.Module):
+
+    def __init__(self, tied):
+        super().__init__()
+        self.config = type("Config", (), {"tie_word_embeddings": not tied})()
+        self.embed_tokens = nn.Embedding(100, 32)
+        self.lm_head = nn.Linear(32, 100, bias=False)
+        if tied:
+            self.lm_head.weight = self.embed_tokens.weight
+
+
 def _build_config():
     """Partition config that matches q_proj and o_proj via regex."""
     return AutoTPConfig(layer_specs=[
@@ -123,5 +135,55 @@ def test_nested_depth_correct():
         assert f"layers.{layer_idx}.self_attn.o_proj" in matched_names
 
 
+def _build_gathered_lm_head_autotp(model, mp_size=1):
+    config = AutoTPConfig(layer_specs=[
+        TPLayerSpec(
+            patterns=[r".*lm_head\.weight$"],
+            partition_type=PartitionType.COLUMN,
+            gather_output=True,
+        ),
+    ])
+    autotp = AutoTP(
+        module=model,
+        all_reduce_linears=[],
+        prefix="",
+        state_dict=None,
+        linear_layer_setting=None,
+        orig_layer_impl=None,
+        partition_config=config,
+    )
+    autotp.set_tensor_parallel_config(mp_size, None)
+    autotp.update_linear_policies()
+    return autotp
+
+
+def test_gathered_lm_head_uses_column_parallel_layer_when_untied():
+    model = OutputModel(tied=False)
+    _build_gathered_lm_head_autotp(model)._replace_module(model)
+
+    assert isinstance(model.lm_head, LinearLayer)
+    assert model.lm_head.gather_output
+
+
+def test_gathered_lm_head_falls_back_for_runtime_parameter_tie():
+    model = OutputModel(tied=True)
+    assert model.lm_head.weight is model.embed_tokens.weight
+
+    _build_gathered_lm_head_autotp(model)._replace_module(model)
+
+    assert isinstance(model.embed_tokens, nn.Embedding)
+    assert isinstance(model.lm_head, nn.Linear)
+    assert model.lm_head.weight is model.embed_tokens.weight
+
+
+def test_gathered_lm_head_falls_back_to_legacy_allreduce_when_output_dim_is_uneven():
+    model = OutputModel(tied=False)
+    model.lm_head = nn.Linear(32, 101, bias=False)
+
+    _build_gathered_lm_head_autotp(model, mp_size=2)._replace_module(model)
+
+    assert isinstance(model.lm_head, LmHeadLinearAllreduce)
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
diff --git a/tests/unit/module_inject/test_tp_plan_converter.py b/tests/unit/module_inject/test_tp_plan_converter.py
index da787153943f..f04b5be00019 100644
--- a/tests/unit/module_inject/test_tp_plan_converter.py
+++ b/tests/unit/module_inject/test_tp_plan_converter.py
@@ -4,7 +4,7 @@
 # DeepSpeed Team
 
 from deepspeed.module_inject.tp_plan_converter import TPPlanConverter
-from deepspeed.module_inject.autotp_config import PartitionType
+from deepspeed.module_inject.autotp_config import AutoTPConfig, PartitionType
 
 
 class TestTPPlanConverter:
@@ -28,7 +28,31 @@ def test_colwise_rowwise_conversion(self):
         o_spec = [s for s in specs if "o_proj" in s.patterns[0]][0]
 
         assert q_spec.partition_type == PartitionType.COLUMN
+        assert not q_spec.gather_output
         assert o_spec.partition_type == PartitionType.ROW
+        assert not o_spec.gather_output
+
+    def test_gathered_colwise_conversion(self):
+        hf_plan = {
+            "lm_head": "colwise_gather_output",
+            "legacy_lm_head": "colwise_rep",
+        }
+        specs = TPPlanConverter.convert(hf_plan)
+
+        assert len(specs) == 2
+        assert all(spec.partition_type == PartitionType.COLUMN for spec in specs)
+        assert all(spec.gather_output for spec in specs)
+
+    def test_gather_output_from_config_dict(self):
+        config = AutoTPConfig.from_dict({
+            "layer_specs": [{
+                "patterns": [r".*\.lm_head\.weight$"],
+                "partition_type": "column",
+                "gather_output": True,
+            }]
+        })
+
+        assert config.layer_specs[0].gather_output
 
     def test_pattern_weight_suffix(self):
         hf_plan = {"layers.*.q_proj": "colwise"}
@@ -87,7 +111,7 @@ def test_pattern_matches_param_name(self):
 
     def test_unsupported_style_returns_none(self):
         """Unsupported styles cause convert() to return None for fallback."""
-        hf_plan = {"layers.*.q_proj": "colwise_rep", "layers.*.o_proj": "rowwise"}
+        hf_plan = {"layers.*.q_proj": "local_colwise", "layers.*.o_proj": "rowwise"}
         result = TPPlanConverter.convert(hf_plan)
         assert result is None
 
diff --git a/tests/unit/runtime/test_tp_plan_extraction.py b/tests/unit/runtime/test_tp_plan_extraction.py
index 7d0c8e81164c..6dff39bbb1a6 100644
--- a/tests/unit/runtime/test_tp_plan_extraction.py
+++ b/tests/unit/runtime/test_tp_plan_extraction.py
@@ -90,7 +90,7 @@ def __init__(self):
 
         assert tp_plan is None
 
-    def test_priority_config_over_model(self):
+    def test_unique_runtime_entries_are_merged_with_config(self):
 
         class MockHFConfig:
             base_model_tp_plan = {"config_plan": "colwise"}
@@ -107,4 +107,57 @@ def __init__(self, config):
 
         assert tp_plan is not None
         assert "config_plan" in tp_plan
-        assert "model_plan" not in tp_plan
+        assert "model_plan" in tp_plan
+
+    def test_prefixed_runtime_duplicates_are_ignored(self):
+
+        class MockHFConfig:
+            base_model_tp_plan = {"layers.*.self_attn.q_proj": "colwise"}
+
+        class MockHFModel:
+
+            def __init__(self, config):
+                self.config = config
+                self._tp_plan = {
+                    "model.layers.*.self_attn.q_proj": "colwise",
+                    "lm_head": "colwise_gather_output",
+                }
+
+        tp_plan = _get_hf_tp_plan(MockHFModel(MockHFConfig()))
+
+        assert tp_plan == {
+            "layers.*.self_attn.q_proj": "colwise",
+            "lm_head": "colwise_gather_output",
+        }
+
+    def test_class_lm_head_plan_survives_runtime_base_plan_override(self):
+
+        class MockHFConfig:
+            base_model_tp_plan = {"layers.*.self_attn.q_proj": "colwise"}
+
+        class MockHFModel:
+            _tp_plan = {"lm_head": "colwise_gather_output"}
+
+            def __init__(self, config):
+                self.config = config
+                self._tp_plan = {
+                    "layers.*.self_attn.q_proj": "colwise",
+                    "model.layers.*.self_attn.q_proj": "colwise",
+                }
+
+        model = MockHFModel(MockHFConfig())
+        assert model._tp_plan != type(model)._tp_plan
+
+        tp_plan = _get_hf_tp_plan(model)
+
+        assert tp_plan == {
+            "layers.*.self_attn.q_proj": "colwise",
+            "lm_head": "colwise_gather_output",
+        }
+
+    def test_extract_class_only_tp_plan(self):
+
+        class MockHFModel:
+            _tp_plan = {"lm_head": "colwise_rep"}
+
+        assert _get_hf_tp_plan(MockHFModel()) == {"lm_head": "colwise_rep"}
diff --git a/tests/unit/v1/moe/test_autoep_autotp_runtime.py b/tests/unit/v1/moe/test_autoep_autotp_runtime.py
index 040e0329c70f..132c54e2adff 100644
--- a/tests/unit/v1/moe/test_autoep_autotp_runtime.py
+++ b/tests/unit/v1/moe/test_autoep_autotp_runtime.py
@@ -103,12 +103,27 @@ def __init__(self):
 
     model = nn.Module()
     model.moe = AutoEPLike()
-    autotp = AutoTP.__new__(AutoTP)
-    calls = []
-    monkeypatch.setattr(autotp, "_replace_autoep_shared_experts", lambda child, name: calls.append((child, name)))
 
-    AutoTP._replace_module(autotp, model)
+    autotp = AutoTP(
+        module=model,
+        all_reduce_linears=[],
+        prefix="",
+        state_dict=None,
+        linear_layer_setting=None,
+        orig_layer_impl=None,
+        partition_config=None,
+    )
+
+    calls = []
+    monkeypatch.setattr(
+        autotp,
+        "_replace_autoep_shared_experts",
+        lambda child, name: calls.append((child, name)),
+    )
 
+    result = autotp._replace_module(model)
+ 
+    assert result is model
     assert calls == [(model.moe, "moe")]