diff --git a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/__init__.py b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/__init__.py
index 1bd55b7d995..47db6401d8e 100644
--- a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/__init__.py
+++ b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/__init__.py
@@ -15,6 +15,19 @@
 """Experimental Megatron-FSDP implementation."""
 
 from .dbuffer import DBuffer
-from .placement import Flat, Partial, Placement, Replicate
+from .fsdp_module import FsdpModule
+from .fully_shard import fully_shard
+from .parameter_group import ParameterGroup
+from .placement import Flat, Partial, Placement, Placements, Replicate
 
-__all__ = ["DBuffer", "Flat", "Partial", "Placement", "Replicate"]
+__all__ = [
+    "DBuffer",
+    "Flat",
+    "FsdpModule",
+    "ParameterGroup",
+    "Partial",
+    "Placement",
+    "Placements",
+    "Replicate",
+    "fully_shard",
+]
diff --git a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/dbuffer.py b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/dbuffer.py
index 51f52451089..5305da55994 100644
--- a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/dbuffer.py
+++ b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/dbuffer.py
@@ -42,6 +42,7 @@ def _validate_mesh_axis(mesh: DeviceMesh, axis: int) -> None:
 
 
 def _validate_placements(placements: Iterable[Placement]) -> None:
+    """Validate DBuffer placements form a supported contiguous local layout."""
     seen_flat = False
     for placement in placements:
         if not isinstance(placement, (Replicate, Partial, Flat)):
@@ -116,6 +117,20 @@ def device(self) -> torch.device:
         """Device of the local buffer."""
         return self.local_buffer.device
 
+    def reallocate_storage(self) -> None:
+        """Restore the local buffer's backing storage to its logical size."""
+        self._resize_storage(self.local_buffer.numel())
+
+    def release_storage(self) -> None:
+        """Release local buffer storage without replacing the Storage object."""
+        # Autograd may save views that share this Storage object. Resizing the
+        # existing Storage releases the allocation while preserving those aliases
+        # for a later reallocate_storage().
+        self._resize_storage(0)
+
+    def _resize_storage(self, numel: int) -> None:
+        self.local_buffer.untyped_storage().resize_(numel * self.local_buffer.element_size())
+
     def _get_owned_range(self, tensor_index: int) -> _OwnedRange | None:
         """Return this buffer's owned range for logical tensor ``tensor_index``."""
         tensor_start = self.layout.tensor_to_offset[tensor_index]
@@ -254,6 +269,21 @@ def _create_or_validate_out(
             raise ValueError(f"Expected out device {self.device}, got {out.device}.")
         return out
 
+    def cast(self, dtype: torch.dtype) -> "DBuffer":
+        """Return this buffer with the same layout and placements in ``dtype``."""
+        if self.dtype == dtype:
+            return self
+
+        destination = DBuffer(
+            mesh=self.mesh,
+            placements=self.placements,
+            tensor_shapes=self.layout.tensor_shapes,
+            dtype=dtype,
+            device=self.device,
+        )
+        destination.local_buffer.copy_(self.local_buffer)
+        return destination
+
     def redistribute(
         self, new_placements: Iterable[Placement], *, out: "DBuffer | None" = None
     ) -> "DBuffer":
diff --git a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/fsdp_module.py b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/fsdp_module.py
new file mode 100644
index 00000000000..5360b916169
--- /dev/null
+++ b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/fsdp_module.py
@@ -0,0 +1,161 @@
+# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Module mixin for the minimal Megatron-FSDP path."""
+
+from collections.abc import Callable
+from typing import cast
+
+import torch
+from torch import nn
+from torch.distributed import DeviceMesh
+
+from ..mixed_precision import MixedPrecisionPolicy
+from .parameter_group import ParameterGroup, contained_in_parameter_group
+from .placement import MeshAxis, Placements
+
+
+class FsdpModule:
+    """Mixin attached to modules managed by the minimal FSDP path."""
+
+    _parameter_groups: tuple[ParameterGroup, ...]
+    _ready_grad_parameters: set[nn.Parameter]
+    _num_training_parameters: int
+
+    def __init__(
+        self, mesh: DeviceMesh, placements: Placements, mixed_precision_policy: MixedPrecisionPolicy
+    ) -> None:
+        """Initialize FSDP runtime state on an already-constructed module."""
+        owned_parameters = _collect_owned_parameters(self)
+        axis_indices = tuple(_axis_index(mesh, axis) for axis in placements.dp_axes)
+        assert axis_indices == tuple(
+            range(mesh.ndim)
+        ), "FSDP requires dp_axes to match every mesh axis in mesh order for now."
+        parameter_groups = [
+            ParameterGroup(
+                owning_module=self,
+                parameters=group_parameters,
+                mesh=mesh,
+                placements=placements,
+                mixed_precision_policy=mixed_precision_policy,
+            )
+            for group_parameters in _group_parameters(owned_parameters)
+        ]
+        self._parameter_groups = tuple(parameter_groups)
+        self._ready_grad_parameters = set()
+        self._num_training_parameters = sum(
+            len(group.sharded_parameters) for group in self._parameter_groups if group.requires_grad
+        )
+        self._register_hooks()
+
+    def _register_hooks(self) -> None:
+        module = cast(nn.Module, self)
+        module.register_forward_pre_hook(lambda _module, _args: self.pre_forward())
+        module.register_forward_hook(lambda _module, _args, _output: self.post_forward())
+        module.register_full_backward_pre_hook(lambda _module, _grad_output: self.pre_backward())
+        # Gradient reduction is parameter-completion based: once every owned
+        # Parameter has accumulated its grad, this FSDP unit can reduce and
+        # reshard. Module full-backward hooks can fire before that when module
+        # inputs do not require grad.
+        for group in self._parameter_groups:
+            if not group.requires_grad:
+                continue
+            for parameter in group.unsharded_parameters:
+                parameter.register_post_accumulate_grad_hook(self._make_grad_hook(parameter))
+
+    def _make_grad_hook(self, parameter: nn.Parameter) -> Callable[[nn.Parameter], None]:
+        def grad_hook(_parameter: nn.Parameter) -> None:
+            self._ready_grad_parameters.add(parameter)
+            if len(self._ready_grad_parameters) == self._num_training_parameters:
+                self.post_backward()
+
+        return grad_hook
+
+    def pre_forward(self) -> None:
+        """Prepare full parameters for forward compute."""
+        self._ready_grad_parameters.clear()
+        for group in self._parameter_groups:
+            group.sync_model_weight_from_main_weight()
+            group.unshard_parameters()
+
+    def post_forward(self) -> None:
+        """Return parameters to their sharded resting state after forward compute."""
+        for group in self._parameter_groups:
+            group.reshard_parameters()
+
+    def pre_backward(self) -> None:
+        """Prepare full parameters for backward compute."""
+        for group in self._parameter_groups:
+            group.unshard_parameters()
+
+    def post_backward(self) -> None:
+        """Reduce gradients and return parameters to their sharded resting state."""
+        for group in self._parameter_groups:
+            if group.requires_grad:
+                group.reduce_gradients()
+            group.reshard_parameters()
+        self._ready_grad_parameters.clear()
+
+    def parameter_groups(self) -> tuple[ParameterGroup, ...]:
+        """Return parameter groups owned by this FSDP unit."""
+        return self._parameter_groups
+
+
+def _axis_index(mesh: DeviceMesh, axis: MeshAxis) -> int:
+    if isinstance(axis, int):
+        axis_index = axis
+        if axis_index < 0:
+            axis_index += mesh.ndim
+        if axis_index < 0 or axis_index >= mesh.ndim:
+            raise ValueError(f"Mesh axis {axis} is out of bounds for mesh ndim {mesh.ndim}.")
+        return axis_index
+
+    dim_names = mesh.mesh_dim_names
+    if dim_names is None or axis not in dim_names:
+        raise ValueError(f"Mesh axis {axis!r} is not present in mesh dim names {dim_names}.")
+    return dim_names.index(axis)
+
+
+def _collect_owned_parameters(root_module: nn.Module) -> dict[str, nn.Parameter]:
+    parameters: dict[str, nn.Parameter] = {}
+
+    def visit(submodule: nn.Module, submodule_fqn: str) -> None:
+        direct_parameters = list(submodule.named_parameters(recurse=False))
+
+        for local_parameter_name, parameter in direct_parameters:
+            parameter_fqn = (
+                f"{submodule_fqn}.{local_parameter_name}" if submodule_fqn else local_parameter_name
+            )
+            if contained_in_parameter_group(parameter):
+                raise ValueError(f"Parameter {parameter_fqn!r} is already owned by an FSDP unit.")
+            parameters[parameter_fqn] = parameter
+
+        for child_name, child_module in submodule.named_children():
+            if isinstance(child_module, FsdpModule):
+                continue
+            child_fqn = f"{submodule_fqn}.{child_name}" if submodule_fqn else child_name
+            visit(child_module, child_fqn)
+
+    visit(root_module, "")
+    if not parameters:
+        raise ValueError("fully_shard requires at least one unowned parameter.")
+    return parameters
+
+
+def _group_parameters(parameters: dict[str, nn.Parameter]) -> list[dict[str, nn.Parameter]]:
+    grouped: dict[tuple[torch.dtype, bool], dict[str, nn.Parameter]] = {}
+    for name, parameter in parameters.items():
+        key = (parameter.dtype, parameter.requires_grad)
+        grouped.setdefault(key, {})[name] = parameter
+    return [grouped[key] for key in grouped]
diff --git a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/fully_shard.py b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/fully_shard.py
new file mode 100644
index 00000000000..3ab3c0e6d71
--- /dev/null
+++ b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/fully_shard.py
@@ -0,0 +1,61 @@
+# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Minimal Megatron-FSDP fully_shard entrypoint."""
+
+from torch import nn
+from torch.distributed import DeviceMesh
+
+from ..mixed_precision import MixedPrecisionPolicy
+from .fsdp_module import FsdpModule
+from .placement import Placements
+
+
+def fully_shard(
+    module: nn.Module,
+    mesh: DeviceMesh,
+    placements: Placements,
+    mixed_precision_policy: MixedPrecisionPolicy | None = None,
+) -> None:
+    """Shard one module as a per-module FSDP unit.
+
+    Args:
+        module: Module whose currently unowned parameters become this FSDP unit.
+        mesh: Device mesh used for sharding.
+        placements: Parameter, gradient, and optimizer placements.
+        mixed_precision_policy: Optional precision policy. Defaults to FP32 main weights
+            and parameter-dtype main gradients.
+    """
+    if isinstance(module, FsdpModule):
+        raise ValueError("This module is already managed by FSDP.")
+
+    mixed_precision_policy = mixed_precision_policy or MixedPrecisionPolicy()
+    original_cls = module.__class__
+    _attach_mixin(module)
+    try:
+        assert isinstance(module, FsdpModule)
+        FsdpModule.__init__(
+            module, mesh=mesh, placements=placements, mixed_precision_policy=mixed_precision_policy
+        )
+    except Exception:
+        module.__class__ = original_cls
+        raise
+
+
+def _attach_mixin(module: nn.Module) -> None:
+    if isinstance(module, FsdpModule):
+        return
+    module_cls = module.__class__
+    fsdp_cls = type(f"ExperimentalFsdp{module_cls.__name__}", (FsdpModule, module_cls), {})
+    module.__class__ = fsdp_cls
diff --git a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/parameter_group.py b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/parameter_group.py
new file mode 100644
index 00000000000..d5cdde1f23e
--- /dev/null
+++ b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/parameter_group.py
@@ -0,0 +1,258 @@
+# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Parameter-group runtime state for the minimal Megatron-FSDP path."""
+
+from collections.abc import Iterable
+
+import torch
+import torch.distributed as dist
+from torch import nn
+from torch.distributed import DeviceMesh
+
+from ..mixed_precision import MixedPrecisionPolicy
+from .dbuffer import DBuffer
+from .placement import Partial, Placements, Replicate
+
+_CONTAINING_PARAMETER_GROUP_ATTR = "_mfsdp_parameter_group"
+
+
+def contained_in_parameter_group(parameter: nn.Parameter) -> bool:
+    """Return whether a parameter is already owned by a ParameterGroup."""
+    return hasattr(parameter, _CONTAINING_PARAMETER_GROUP_ATTR)
+
+
+class ParameterGroup:
+    """A dtype and requires-grad homogeneous group of FSDP-owned parameters."""
+
+    owning_module: nn.Module
+    parameter_names: tuple[str, ...]
+    sharded_parameters: tuple[nn.Parameter, ...]
+    unsharded_parameters: tuple[nn.Parameter, ...]
+    mesh: DeviceMesh
+    dtype: torch.dtype
+    requires_grad: bool
+    main_weight: DBuffer
+    model_weight: DBuffer
+    main_grad: DBuffer | None
+    _unsharded_model_weight: DBuffer
+
+    def __init__(
+        self,
+        owning_module: nn.Module,
+        parameters: dict[str, nn.Parameter],
+        mesh: DeviceMesh,
+        placements: Placements,
+        mixed_precision_policy: MixedPrecisionPolicy,
+    ) -> None:
+        """Create persistent sharded buffers for a group of parameters.
+
+        Args:
+            owning_module: Closest FSDP root module that owns this parameter group.
+            parameters: Root-module-relative FQNs and their parameters.
+            mesh: Device mesh used for all DBuffer storage in this version.
+            placements: Parameter, gradient, and optimizer placements.
+            mixed_precision_policy: Precision policy for main weights and gradients.
+        """
+        if not parameters:
+            raise ValueError("ParameterGroup requires at least one parameter.")
+
+        model_weight_placements = tuple(placements.parameter)
+        main_grad_placements = tuple(placements.gradient)
+        main_weight_placements = tuple(placements.optimizer)
+
+        # Python dicts preserve insertion order, so parameter_names and
+        # parameters.values() define the same stable DBuffer tensor order.
+        self.owning_module = owning_module
+        self.mesh = mesh
+        self.parameter_names = tuple(parameters)
+        first_parameter = next(iter(parameters.values()))
+        self.dtype = first_parameter.dtype
+        self.requires_grad = first_parameter.requires_grad
+        for name, parameter in parameters.items():
+            if parameter.dtype != self.dtype:
+                raise ValueError(
+                    f"Expected parameter {name!r} to have dtype {self.dtype}, "
+                    f"got {parameter.dtype}."
+                )
+            if parameter.requires_grad != self.requires_grad:
+                raise ValueError(
+                    f"Expected parameter {name!r} to have requires_grad={self.requires_grad}, "
+                    f"got {parameter.requires_grad}."
+                )
+
+        tensor_shapes = tuple(parameter.shape for parameter in parameters.values())
+        main_weight_dtype = mixed_precision_policy.main_params_dtype or torch.float32
+        self.main_weight = DBuffer.distribute_tensors(
+            (parameter.to(dtype=main_weight_dtype) for parameter in parameters.values()),
+            mesh=self.mesh,
+            placements=main_weight_placements,
+        )
+
+        self._unsharded_model_weight = DBuffer(
+            mesh=self.mesh,
+            placements=[Replicate()] * self.mesh.ndim,
+            tensor_shapes=tensor_shapes,
+            dtype=self.dtype,
+            device=self.main_weight.device,
+        )
+        if main_weight_dtype == self.dtype and main_weight_placements == model_weight_placements:
+            self.model_weight = self.main_weight
+        else:
+            self.model_weight = DBuffer(
+                mesh=self.mesh,
+                placements=model_weight_placements,
+                tensor_shapes=tensor_shapes,
+                dtype=self.dtype,
+                device=self.main_weight.device,
+            )
+
+        self.main_grad = None
+        if self.requires_grad:
+            grad_dtype = mixed_precision_policy.main_grads_dtype or self.dtype
+            self.main_grad = DBuffer(
+                mesh=self.mesh,
+                placements=main_grad_placements,
+                tensor_shapes=self.main_weight.layout.tensor_shapes,
+                dtype=grad_dtype,
+                device=self.main_weight.device,
+            )
+            assert self.main_grad.layout == self.main_weight.layout, (
+                "main_grad is built from main_weight tensor shapes on the same mesh, "
+                "and DBuffer layouts are deterministic from those shapes and mesh size."
+            )
+            if self.main_grad.placements != self.main_weight.placements:
+                raise ValueError(
+                    "FSDP temporarily requires main_grad and main_weight to have the same "
+                    "placements until HSDP/HFSDP support is implemented. "
+                    f"Got main_grad placements {self.main_grad.placements} and "
+                    f"main_weight placements {self.main_weight.placements}."
+                )
+
+        sharded_parameters: list[nn.Parameter] = []
+        unsharded_parameters: list[nn.Parameter] = []
+        main_grad_dtype = self.main_grad.dtype if self.main_grad is not None else None
+        for index, parameter in enumerate(parameters.values()):
+            parameter.data = self._unsharded_model_weight.get_local_tensor(index)
+            parameter.grad = None
+            setattr(parameter, _CONTAINING_PARAMETER_GROUP_ATTR, self)
+            unsharded_parameters.append(parameter)
+
+            sharded_parameter = nn.Parameter(
+                self.main_weight.get_dtensor(index), requires_grad=parameter.requires_grad
+            )
+            if main_grad_dtype:
+                sharded_parameter.grad_dtype = main_grad_dtype
+            setattr(sharded_parameter, _CONTAINING_PARAMETER_GROUP_ATTR, self)
+            sharded_parameters.append(sharded_parameter)
+        self.sharded_parameters = tuple(sharded_parameters)
+        self.unsharded_parameters = tuple(unsharded_parameters)
+
+        self._switch_to_sharded_parameters()
+        self._unsharded_model_weight.release_storage()
+
+    def _set_module_parameters(self, parameters: tuple[nn.Parameter, ...]) -> None:
+        for name, parameter in zip(self.parameter_names, parameters, strict=True):
+            module, parameter_name = _get_parameter_owner(self.owning_module, name)
+            module._parameters[parameter_name] = parameter
+
+    def _switch_to_sharded_parameters(self) -> None:
+        self._set_module_parameters(self.sharded_parameters)
+
+    def _switch_to_unsharded_parameters(self) -> None:
+        self._set_module_parameters(self.unsharded_parameters)
+
+    def sync_model_weight_from_main_weight(self) -> None:
+        """Refresh compute weights from optimizer weights."""
+        if self.main_weight is self.model_weight:
+            return
+
+        self.main_weight.cast(self.model_weight.dtype).redistribute(
+            self.model_weight.placements, out=self.model_weight
+        )
+
+    def unshard_parameters(self) -> None:
+        """Install full parameters for local compute."""
+        self._unsharded_model_weight.reallocate_storage()
+        # This buffer backs unsharded Parameters whose views may be saved by autograd.
+        # Materializing FSDP-managed storage should not look like a user mutation.
+        with torch.autograd._unsafe_preserve_version_counter(
+            self._unsharded_model_weight.local_buffer
+        ):
+            self.model_weight.redistribute(
+                self._unsharded_model_weight.placements, out=self._unsharded_model_weight
+            )
+        self._switch_to_unsharded_parameters()
+
+    def reshard_parameters(self) -> None:
+        """Install sharded DTensor parameters on the owning modules."""
+        self._switch_to_sharded_parameters()
+        self._unsharded_model_weight.release_storage()
+
+    def reduce_gradients(self) -> None:
+        """Reduce full local gradients into sharded parameter gradients."""
+        assert self.main_grad is not None
+
+        def has_grad(parameters: Iterable[nn.Parameter]) -> bool:
+            has_any_grad = False
+            has_any_missing_grad = False
+            for parameter in parameters:
+                if parameter.grad is None:
+                    has_any_missing_grad = True
+                else:
+                    has_any_grad = True
+            if has_any_grad and has_any_missing_grad:
+                raise RuntimeError("FSDP sharded gradients must be either all set or all None.")
+            return has_any_grad
+
+        grads: list[torch.Tensor] = []
+        for name, parameter in zip(self.parameter_names, self.unsharded_parameters, strict=True):
+            if parameter.grad is None:
+                raise RuntimeError(f"Missing gradient for FSDP parameter {name!r}.")
+            grads.append(parameter.grad)
+
+        partial_grad = DBuffer.distribute_tensors(
+            grads, mesh=self.mesh, placements=[Partial(dist.ReduceOp.AVG)] * self.mesh.ndim
+        )
+
+        # zero_grad(set_to_none=True) clears sharded parameter grads, so the next
+        # backward can reduce directly into main_grad. zero_grad(set_to_none=False)
+        # leaves sharded grads installed, so this backward accumulates into main_grad.
+        has_sharded_grads = has_grad(self.sharded_parameters)
+        can_reduce_into_main_grad = (
+            not has_sharded_grads
+            and partial_grad.dtype == self.main_grad.dtype
+        )
+        if can_reduce_into_main_grad:
+            partial_grad.redistribute(self.main_grad.placements, out=self.main_grad)
+        else:
+            reduced_grad = partial_grad.redistribute(self.main_grad.placements)
+            if has_sharded_grads:
+                self.main_grad.local_buffer.add_(reduced_grad.local_buffer)
+            else:
+                self.main_grad.local_buffer.copy_(reduced_grad.local_buffer)
+
+        if not has_sharded_grads:
+            for index, parameter in enumerate(self.sharded_parameters):
+                parameter.grad = self.main_grad.get_dtensor(index)
+
+        for parameter in self.unsharded_parameters:
+            parameter.grad = None
+
+
+def _get_parameter_owner(module: nn.Module, name: str) -> tuple[nn.Module, str]:
+    """Resolve a root-module-relative parameter FQN to its direct owner."""
+    module_name, separator, parameter_name = name.rpartition(".")
+    owner = module.get_submodule(module_name) if separator else module
+    return owner, parameter_name
diff --git a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/placement.py b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/placement.py
index 1b561c9634d..5e4dc6b985e 100644
--- a/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/placement.py
+++ b/megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/placement.py
@@ -38,6 +38,9 @@ class Placement:
     """Base class for DBuffer placements."""
 
 
+MeshAxis = int | str
+
+
 @dataclasses.dataclass(frozen=True)
 class Replicate(Placement):
     """Replicated local buffer placement."""
@@ -53,3 +56,24 @@ class Partial(Placement):
 @dataclasses.dataclass(frozen=True)
 class Flat(Placement):
     """Flat per-unit dim-0 sharded local buffer placement."""
+
+
+@dataclasses.dataclass(frozen=True)
+class Placements:
+    """Per-mesh-axis placements for parameter, gradient, and optimizer buffers."""
+
+    dp_axes: list[MeshAxis]
+    parameter: list[Placement]
+    gradient: list[Placement]
+    optimizer: list[Placement]
+
+    def __post_init__(self) -> None:
+        """Validate placement list lengths."""
+        axis_count = len(self.dp_axes)
+        for name, placements in (
+            ("parameter", self.parameter),
+            ("gradient", self.gradient),
+            ("optimizer", self.optimizer),
+        ):
+            if len(placements) != axis_count:
+                raise ValueError(f"Expected {axis_count} {name} placements, got {len(placements)}.")
diff --git a/tests/unit_tests/distributed/megatron_fsdp/test_dbuffer.py b/tests/unit_tests/distributed/megatron_fsdp/test_dbuffer.py
index a6ae56dc852..d267298b94d 100644
--- a/tests/unit_tests/distributed/megatron_fsdp/test_dbuffer.py
+++ b/tests/unit_tests/distributed/megatron_fsdp/test_dbuffer.py
@@ -155,6 +155,65 @@ def test_constructor_allocates_local_buffer(distributed_setup):
     assert sharded_buffer.local_buffer.device == distributed_setup.device
 
 
+@pytest.mark.distributed
+def test_cast_to_same_dtype_returns_self(distributed_setup):
+    """DBuffer.cast returns self when the dtype already matches."""
+    mesh = init_device_mesh(distributed_setup.device.type, (distributed_setup.world_size,))
+    tensors = _same_tensors_on_all_ranks(distributed_setup.device)
+    buffer = DBuffer.distribute_tensors(tensors, mesh, [Replicate()])
+
+    assert buffer.cast(torch.float32) is buffer
+
+
+@pytest.mark.distributed
+def test_cast_preserves_layout_and_casts_values(distributed_setup):
+    """DBuffer.cast preserves layout metadata and casts local values."""
+    mesh = init_device_mesh(distributed_setup.device.type, (distributed_setup.world_size,))
+    tensors = _same_tensors_on_all_ranks(distributed_setup.device)
+    buffer = DBuffer.distribute_tensors(tensors, mesh, [Replicate()])
+
+    cast_buffer = buffer.cast(torch.bfloat16)
+
+    assert cast_buffer is not buffer
+    assert cast_buffer.mesh == buffer.mesh
+    assert cast_buffer.placements == buffer.placements
+    assert cast_buffer.layout == buffer.layout
+    assert cast_buffer.device == buffer.device
+    assert cast_buffer.dtype is torch.bfloat16
+    _assert_dbuffer_local_tensors_close(
+        cast_buffer, [tensor.to(dtype=torch.bfloat16) for tensor in tensors]
+    )
+
+
+@pytest.mark.distributed
+def test_release_and_reallocate_storage_preserves_buffer_views(distributed_setup):
+    """DBuffer storage can be released and reallocated without replacing existing views."""
+    mesh = init_device_mesh(distributed_setup.device.type, (distributed_setup.world_size,))
+    buffer = DBuffer(
+        mesh=mesh,
+        placements=[Replicate()],
+        tensor_shapes=[torch.Size((4, 4))],
+        dtype=torch.float32,
+        device=distributed_setup.device,
+    )
+    tensor_view = buffer.get_local_tensor(0)
+    buffer_data_ptr = buffer.local_buffer.data_ptr()
+    tensor_view_data_ptr = tensor_view.data_ptr()
+
+    buffer.release_storage()
+    assert buffer.local_buffer.untyped_storage().nbytes() == 0
+
+    buffer.reallocate_storage()
+    assert (
+        buffer.local_buffer.untyped_storage().nbytes()
+        == buffer.local_buffer.numel() * buffer.local_buffer.element_size()
+    )
+    assert buffer.local_buffer.data_ptr() == buffer_data_ptr
+    assert tensor_view.data_ptr() == tensor_view_data_ptr
+    buffer.local_buffer.fill_(7.0)
+    torch.testing.assert_close(tensor_view, torch.full_like(tensor_view, 7.0))
+
+
 @pytest.mark.distributed
 def test_from_local_reuses_required_local_buffer(distributed_setup):
     """DBuffer.from_local reuses caller-provided local storage without allocation."""
@@ -203,6 +262,24 @@ def test_distribute_tensors_moves_inputs_to_mesh_device(distributed_setup):
     )
 
 
+@pytest.mark.distributed
+def test_distribute_tensors_detaches_and_contiguizes_inputs(distributed_setup):
+    """distribute_tensors treats input tensors as detached contiguous values."""
+    mesh = init_device_mesh(distributed_setup.device.type, (distributed_setup.world_size,))
+    parameter = torch.nn.Parameter(
+        torch.arange(12, dtype=torch.float32, device=distributed_setup.device).view(3, 4).t()
+    )
+
+    buffer = DBuffer.distribute_tensors([parameter], mesh, [Replicate()])
+
+    assert not parameter.is_contiguous()
+    assert buffer.get_local_tensor(0).is_contiguous()
+    assert not buffer.local_buffer.requires_grad
+    torch.testing.assert_close(
+        buffer.get_local_tensor(0), parameter.detach().contiguous(), rtol=0, atol=0
+    )
+
+
 @pytest.mark.distributed
 def test_sharded_allgather_round_trip(distributed_setup):
     """Sharded buffers round-trip through all-gather as contiguous tensor fragments."""
diff --git a/tests/unit_tests/distributed/megatron_fsdp/test_experimental_fully_shard.py b/tests/unit_tests/distributed/megatron_fsdp/test_experimental_fully_shard.py
new file mode 100644
index 00000000000..957f78a5665
--- /dev/null
+++ b/tests/unit_tests/distributed/megatron_fsdp/test_experimental_fully_shard.py
@@ -0,0 +1,357 @@
+# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+
+"""Unit tests for the minimal Megatron-FSDP path."""
+
+import logging
+
+import pytest
+import torch
+from torch import nn
+from torch.distributed.device_mesh import init_device_mesh
+from torch.distributed.tensor import DTensor
+
+from megatron.core.distributed.fsdp.src.megatron_fsdp.experimental import (
+    Flat,
+    Placements,
+    fully_shard,
+)
+from megatron.core.distributed.fsdp.src.megatron_fsdp.mixed_precision import MixedPrecisionPolicy
+
+logger = logging.getLogger(__name__)
+
+
+class TinyModel(nn.Module):
+    """Small model with two separately shardable units."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.fc1 = nn.Linear(8, 16)
+        self.relu = nn.ReLU()
+        self.fc2 = nn.Linear(16, 4)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """Run the tiny model."""
+        return self.fc2(self.relu(self.fc1(x)))
+
+
+class NestedModel(nn.Module):
+    """Model with direct and child-owned parameters."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.bias = nn.Parameter(torch.ones(4))
+        self.inner = nn.Linear(4, 4, bias=False)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """Run the nested model."""
+        return self.inner(x) + self.bias
+
+
+class SaveNonLeafWeightView(torch.autograd.Function):
+    """Autograd function that saves a non-leaf parameter view for backward."""
+
+    @staticmethod
+    def forward(ctx, x: torch.Tensor, weight_view: torch.Tensor) -> torch.Tensor:
+        """Save the non-leaf weight view and run a simple elementwise op."""
+        ctx.save_for_backward(x, weight_view)
+        return x * weight_view
+
+    @staticmethod
+    def backward(ctx, grad_output: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
+        """Use the saved non-leaf weight view during backward."""
+        x, weight_view = ctx.saved_tensors
+        return grad_output * weight_view, grad_output * x
+
+
+class NonLeafViewModel(nn.Module):
+    """Model that saves a non-leaf parameter view across forward and backward."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.weight = nn.Parameter(torch.randn(8))
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """Run using a non-leaf view of the parameter."""
+        weight_view = self.weight.view_as(self.weight)
+        assert self.weight.is_leaf
+        assert not weight_view.is_leaf
+        return SaveNonLeafWeightView.apply(x, weight_view)
+
+
+def _flat_placements() -> Placements:
+    return Placements(dp_axes=[0], parameter=[Flat()], gradient=[Flat()], optimizer=[Flat()])
+
+
+def _mb(num_bytes: int) -> str:
+    return f"{num_bytes / 1024**2:.2f} MB"
+
+
+@pytest.mark.distributed
+def test_fully_shard_losses_match_baseline(distributed_setup):
+    """Minimal per-module FSDP training should match single-rank SGD."""
+    rank = distributed_setup.rank
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    torch.manual_seed(1234)
+    baseline = TinyModel().to(device)
+    model = TinyModel().to(device)
+    model.load_state_dict(baseline.state_dict())
+
+    fully_shard(model.fc1, mesh=mesh, placements=_flat_placements())
+    fully_shard(model.fc2, mesh=mesh, placements=_flat_placements())
+    baseline_optimizer = torch.optim.SGD(baseline.parameters(), lr=0.05)
+    optimizer = torch.optim.SGD(model.parameters(), lr=0.05)
+
+    x = torch.randn(3, 8, device=device)
+    target = torch.randn(3, 4, device=device)
+
+    for step in range(5):
+        baseline_optimizer.zero_grad()
+        optimizer.zero_grad()
+
+        baseline_loss = torch.nn.functional.mse_loss(baseline(x), target)
+        loss = torch.nn.functional.mse_loss(model(x), target)
+        logger.info(
+            "FSDP train parity: rank=%s, step=%s, baseline_loss=%s, sharded_loss=%s",
+            rank,
+            step,
+            baseline_loss.item(),
+            loss.item(),
+        )
+        torch.testing.assert_close(loss, baseline_loss, msg=f"Loss mismatch at step {step}.")
+
+        baseline_loss.backward()
+        loss.backward()
+        baseline_optimizer.step()
+        optimizer.step()
+
+
+@pytest.mark.distributed
+def test_nested_fully_shard_excludes_child_owned_parameters(distributed_setup):
+    """An outer FSDP unit owns direct parameters but not nested child-unit parameters."""
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    model = NestedModel().to(device)
+
+    fully_shard(model.inner, mesh=mesh, placements=_flat_placements())
+    fully_shard(model, mesh=mesh, placements=_flat_placements())
+
+    inner_names = [
+        name for group in model.inner.parameter_groups() for name in group.parameter_names
+    ]
+    outer_names = [name for group in model.parameter_groups() for name in group.parameter_names]
+
+    assert inner_names == ["weight"]
+    assert outer_names == ["bias"]
+
+
+@pytest.mark.distributed
+def test_frozen_parameter_group_does_not_allocate_main_grad(distributed_setup):
+    """A non-trainable parameter group should not allocate persistent main gradients."""
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    model = nn.Linear(4, 4, bias=False).to(device)
+    model.weight.requires_grad_(False)
+
+    fully_shard(model, mesh=mesh, placements=_flat_placements())
+
+    (group,) = model.parameter_groups()
+    assert not group.requires_grad
+    assert group.main_grad is None
+
+
+pytest.mark.distributed
+
+
+def test_backward_averages_across_dp_and_accumulates_across_calls(distributed_setup):
+    """Each backward averages over DP ranks; repeated backwards accumulate by summing."""
+    rank = distributed_setup.rank
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    model = nn.Linear(1, world_size, bias=False).to(device)
+    with torch.no_grad():
+        model.weight.fill_(1.0)
+
+    fully_shard(model, mesh=mesh, placements=_flat_placements())
+
+    x = torch.full((1, 1), float(rank + 1), device=device)
+    model(x).sum().backward()
+    model(x).sum().backward()
+
+    assert isinstance(model.weight.grad, DTensor)
+    local_grad = model.weight.grad.to_local()
+    expected = torch.full_like(local_grad, float(world_size + 1))
+    torch.testing.assert_close(local_grad, expected, rtol=0, atol=0)
+
+
+@pytest.mark.distributed
+def test_next_forward_uses_optimizer_updated_weights(distributed_setup):
+    """The next forward should observe weights updated by the previous optimizer step."""
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    model = nn.Linear(1, world_size, bias=False, dtype=torch.bfloat16).to(device)
+    with torch.no_grad():
+        model.weight.fill_(1.0)
+
+    fully_shard(
+        model,
+        mesh=mesh,
+        placements=_flat_placements(),
+        mixed_precision_policy=MixedPrecisionPolicy(main_params_dtype=torch.float32),
+    )
+    # SGD's foreach/fused CUDA paths require matching parameter and gradient dtypes.
+    # Use the scalar path to exercise FP32 main weights with default BF16 main grads.
+    optimizer = torch.optim.SGD(model.parameters(), lr=0.25, foreach=False)
+    x = torch.ones(1, 1, device=device, dtype=torch.bfloat16)
+
+    def train_iteration() -> torch.Tensor:
+        optimizer.zero_grad(set_to_none=True)
+        loss = model(x).sum()
+        loss.backward()
+        optimizer.step()
+        return loss.detach().float()
+
+    first_loss = train_iteration()
+    second_loss = train_iteration()
+
+    with pytest.raises(AssertionError):
+        torch.testing.assert_close(second_loss, first_loss)
+
+
+@pytest.mark.distributed
+def test_cpu_initialized_parameters_shard_to_mesh_device(distributed_setup):
+    """CPU-initialized parameters should be sharded with their real values."""
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    model = nn.Linear(4, 4, bias=False)
+    with torch.no_grad():
+        model.weight.fill_(3.0)
+    expected_weight = model.weight.detach().to(device)
+
+    fully_shard(model, mesh=mesh, placements=_flat_placements())
+
+    (group,) = model.parameter_groups()
+    full_weight = group.model_weight.allgather(0).get_local_tensor(0)
+    assert full_weight.device.type == device.type
+    torch.testing.assert_close(full_weight, expected_weight)
+
+
+@pytest.mark.distributed
+def test_non_leaf_parameter_view_survives_storage_resize(distributed_setup):
+    """A non-leaf parameter view saved for backward should survive full-storage resize."""
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    model = NonLeafViewModel().to(device)
+    fully_shard(model, mesh=mesh, placements=_flat_placements())
+
+    group = model.parameter_groups()[0]
+    x = torch.randn(8, device=device, requires_grad=True)
+    loss = model(x).sum()
+
+    assert group._unsharded_model_weight is not None
+    assert group._unsharded_model_weight.local_buffer.untyped_storage().nbytes() == 0
+
+    loss.backward()
+
+    assert group.main_grad is not None
+    assert group._unsharded_model_weight is not None
+    assert group._unsharded_model_weight.local_buffer.untyped_storage().nbytes() == 0
+
+
+@pytest.mark.distributed
+def test_fully_shard_reduces_peak_training_memory(distributed_setup):
+    """Per-layer FSDP should reduce peak CUDA memory during training."""
+    rank = distributed_setup.rank
+    world_size = distributed_setup.world_size
+    device = distributed_setup.device
+    if world_size < 2:
+        pytest.skip("This test requires at least 2 ranks.")
+    if device.type != "cuda":
+        pytest.skip("Peak memory verification requires CUDA.")
+
+    mesh = init_device_mesh(device.type, (world_size,))
+    dim = 1024
+    layers = 16
+    batch = 8
+    steps = 2
+    dtype = torch.bfloat16
+
+    def train_steps(model: nn.Module, optimizer: torch.optim.Optimizer, x: torch.Tensor) -> None:
+        for _ in range(steps):
+            optimizer.zero_grad(set_to_none=True)
+            model(x).sum().backward()
+            optimizer.step()
+
+    torch.manual_seed(4321)
+    baseline = nn.Sequential(*[nn.Linear(dim, dim, dtype=dtype) for _ in range(layers)]).to(
+        device
+    )
+    baseline_optimizer = torch.optim.AdamW(baseline.parameters(), lr=0.01)
+    x = torch.randn(batch, dim, device=device, dtype=dtype)
+    torch.cuda.reset_peak_memory_stats(device)
+    train_steps(baseline, baseline_optimizer, x)
+    torch.cuda.synchronize(device)
+    baseline_peak = torch.cuda.max_memory_allocated(device)
+
+    del baseline_optimizer
+    del baseline
+    del x
+    torch.cuda.empty_cache()
+
+    torch.manual_seed(4321)
+    model = nn.Sequential(*[nn.Linear(dim, dim, dtype=dtype) for _ in range(layers)]).to(
+        device
+    )
+    for layer in model:
+        fully_shard(
+            layer,
+            mesh=mesh,
+            placements=_flat_placements(),
+            mixed_precision_policy=MixedPrecisionPolicy(
+                main_params_dtype=dtype, main_grads_dtype=dtype
+            ),
+        )
+    optimizer = torch.optim.AdamW(model.parameters(), lr=0.01)
+    torch.cuda.empty_cache()
+
+    x = torch.randn(batch, dim, device=device, dtype=dtype)
+    torch.cuda.reset_peak_memory_stats(device)
+    train_steps(model, optimizer, x)
+    torch.cuda.synchronize(device)
+    sharded_peak = torch.cuda.max_memory_allocated(device)
+    logger.info(
+        "FSDP peak memory: rank=%s, baseline=%s, sharded=%s",
+        rank,
+        _mb(baseline_peak),
+        _mb(sharded_peak),
+    )
+
+    assert sharded_peak < baseline_peak