diff --git a/deepspeed/runtime/base_optimizer.py b/deepspeed/runtime/base_optimizer.py
index cf69b62f7d4f..f04c5d35f5bd 100644
--- a/deepspeed/runtime/base_optimizer.py
+++ b/deepspeed/runtime/base_optimizer.py
@@ -252,6 +252,23 @@ class ZeROOptimizer(DeepSpeedOptimizer):
 
     def __init__(self):
         self._backward_hook_state = BackwardHookStateManager()
+        # Mirrored copy of the engine GAS boundary for managed reduce/offload paths.
+        # Engine owns the source of truth (micro-step / step() / set_*); ZeRO reads this
+        # during backward. Prefer get/set methods over touching the private field.
+        self._is_gradient_accumulation_boundary = True
+
+    def is_gradient_accumulation_boundary(self) -> bool:
+        """Whether the current micro-batch is a gradient accumulation boundary.
+
+        Used by managed ZeRO reduce/partition/offload logic. Unmanaged mode still
+        mirrors True while ``engine.step()`` runs so late readers stay consistent;
+        deferred offload finalize does not branch on this flag.
+        """
+        return self._is_gradient_accumulation_boundary
+
+    def set_gradient_accumulation_boundary(self, is_boundary: bool) -> None:
+        """Mirror the engine's gradient accumulation boundary into this optimizer."""
+        self._is_gradient_accumulation_boundary = bool(is_boundary)
 
     # Delegate backward hook state management to the manager.
     # These properties provide backward compatibility with code that accesses
diff --git a/deepspeed/runtime/engine.py b/deepspeed/runtime/engine.py
index 23b28751e2ea..5db39d237967 100755
--- a/deepspeed/runtime/engine.py
+++ b/deepspeed/runtime/engine.py
@@ -1662,12 +1662,6 @@ def _do_sanity_check(self):
                 f'Client Optimizer (type = {type(self.client_optimizer)} is not instantiated but Client LR Scheduler is instantiated'
 
         if not self.managed_gradient_accumulation():
-            offload_optimizer = self.zero_offload_optimizer()
-            offload_param = self.zero_offload_param()
-            assert offload_optimizer is None or offload_optimizer.device == OffloadDeviceEnum.none, \
-                "managed_gradient_accumulation=False is not supported with ZeRO optimizer state offload"
-            assert offload_param is None or offload_param.device == OffloadDeviceEnum.none, \
-                "managed_gradient_accumulation=False is not supported with ZeRO parameter offload"
             assert self.zero_optimization_partition_gradients() or not self.zero_overlap_comm(), \
                 "managed_gradient_accumulation=False supports ZeRO overlap_comm only with ZeRO stage 2"
             assert not self.pipeline_parallelism, \
@@ -2856,7 +2850,8 @@ def allreduce_gradients(self, bucket_size=MEMORY_OPT_ALLREDUCE_SIZE):
             return
 
         # Pass (PP) gas boundary flag to optimizer (required for zero)
-        self.optimizer.is_gradient_accumulation_boundary = self.is_gradient_accumulation_boundary()
+        if hasattr(self.optimizer, "set_gradient_accumulation_boundary"):
+            self.optimizer.set_gradient_accumulation_boundary(self.is_gradient_accumulation_boundary())
         if self.is_gradient_accumulation_boundary():
             self._reduce_autoep_folding_tp_replicated_gradients()
         # ZeRO stage >= 2 communicates during non gradient accumulation boundaries as well
@@ -2923,7 +2918,7 @@ def _backward_prologue(self):
             self.optimizer.zenflow_state ^= 1
 
         if self.zero_optimization():
-            self.optimizer.is_gradient_accumulation_boundary = self.is_gradient_accumulation_boundary()
+            self.optimizer.set_gradient_accumulation_boundary(self.is_gradient_accumulation_boundary())
 
         self._start_timers(self.engine_timers.backward_inner_timers)
 
@@ -3058,7 +3053,7 @@ def coalesce_grad_reduction(self):
             optimizer._coalesce_grad_reduction = False
             self.inside_no_sync_ctxt = False
             self._is_gradient_accumulation_boundary = True
-            optimizer.is_gradient_accumulation_boundary = True
+            optimizer.set_gradient_accumulation_boundary(True)
             try:
                 # Drive a single reduction pass over locally accumulated grads.
                 # Iterate explicitly (rather than calling reduce_gradients) so
@@ -3270,7 +3265,8 @@ def set_gradient_accumulation_boundary(self, is_boundary):
             "set_gradient_accumulation_boundary() is not supported with managed_gradient_accumulation=False; " \
             "the caller owns the boundary by calling step()"
         self._is_gradient_accumulation_boundary = is_boundary
-        self.optimizer.is_gradient_accumulation_boundary = is_boundary
+        if hasattr(self.optimizer, "set_gradient_accumulation_boundary"):
+            self.optimizer.set_gradient_accumulation_boundary(is_boundary)
 
     def zero_grad(self):
         """
@@ -3392,7 +3388,7 @@ def step(self, lr_kwargs=None):
         # Unmanaged mode: step() is the accumulation boundary.
         self._running_engine_step = True
 
-        # Unmanaged boundary: stage 2/3 already reduced/partitioned per backward so only finalize; stage 0/1/DDP reduce here.
+        # Unmanaged boundary: stage 2/3 already reduced/partitioned per backward so only finalize (incl. offload); stage 0/1/DDP reduce here.
         if not self.managed_gradient_accumulation():
             if self.zero_optimization_partition_gradients():
                 self.optimizer.finalize_gradient_accumulation_boundary()
diff --git a/deepspeed/runtime/superoffload/superoffload_stage3.py b/deepspeed/runtime/superoffload/superoffload_stage3.py
index 7c496a3dda37..e048e9ce8ba3 100644
--- a/deepspeed/runtime/superoffload/superoffload_stage3.py
+++ b/deepspeed/runtime/superoffload/superoffload_stage3.py
@@ -151,7 +151,7 @@ def partition_grads(self, params_to_release: List[Parameter], grad_partitions: L
                 grad_buffer.copy_(cuda_grad_buffer, non_blocking=True)
                 grad_buffer = cuda_grad_buffer
 
-            if self.is_gradient_accumulation_boundary:
+            if self.is_gradient_accumulation_boundary():
                 self.norm_for_param_grads[self.get_param_id(param)] = self._constant_buffered_norm2(grad_buffer)
 
                 fp32_grad_tensor = self.fp32_partitioned_groups_flat[i].grad.narrow(
@@ -162,7 +162,7 @@ def partition_grads(self, params_to_release: List[Parameter], grad_partitions: L
             if self.sub_group_grad_partition_counts[i] == self.sub_group_to_param_num[i]:
                 completed_sub_groups.append(i)
 
-        if self.is_gradient_accumulation_boundary and completed_sub_groups:
+        if self.is_gradient_accumulation_boundary() and completed_sub_groups:
             get_accelerator().current_stream().synchronize()
             for i in completed_sub_groups:
                 if self.subgroup_to_device[i] == 'cpu' and not self.clip_grad:
diff --git a/deepspeed/runtime/zenflow/engine_stage3.py b/deepspeed/runtime/zenflow/engine_stage3.py
index c6d749a8fbee..1de39a618964 100644
--- a/deepspeed/runtime/zenflow/engine_stage3.py
+++ b/deepspeed/runtime/zenflow/engine_stage3.py
@@ -325,7 +325,7 @@ def _process_selected_fp32_groups_grad(optimizer_z3, params_to_update, grad_part
 
             curr_buffer_idx += 1
 
-    if not optimizer_z3.is_gradient_accumulation_boundary:
+    if not optimizer_z3.is_gradient_accumulation_boundary():
         optimizer_z3.selective_optimizer.group_step(params_to_update)
     else:
         optimizer_z3.selective_optimizer.temp_copy_param(params_to_update)
@@ -494,7 +494,7 @@ def partition_grads(optimizer_z3, params_to_release: List[Parameter], grad_parti
             i, dest_offset, _ = optimizer_z3.grad_position[optimizer_z3.get_param_id(param)]
             now_state = optimizer_z3.get_overlap_step_state()
 
-            if optimizer_z3.is_gradient_accumulation_boundary:
+            if optimizer_z3.is_gradient_accumulation_boundary():
                 optimizer_z3.norm_for_param_grads[optimizer_z3.get_param_id(
                     param)] = optimizer_z3._constant_buffered_norm2(grad_buffer)
 
diff --git a/deepspeed/runtime/zenflow/zenflow_stage_1_and_2.py b/deepspeed/runtime/zenflow/zenflow_stage_1_and_2.py
index a4f2c249610e..f971e86cfa50 100644
--- a/deepspeed/runtime/zenflow/zenflow_stage_1_and_2.py
+++ b/deepspeed/runtime/zenflow/zenflow_stage_1_and_2.py
@@ -411,7 +411,7 @@ def _process_selected_fp32_groups_grad(self, tensor, total_size, communication_d
 
                 self.param_id_sum_buffer_offset[param_id] = []
 
-        if not self.is_gradient_accumulation_boundary:
+        if not self.is_gradient_accumulation_boundary():
             self.selective_optimizer.group_step(group_to_paramlist)
         else:
             self.selective_optimizer.temp_copy_param(group_to_paramlist)
diff --git a/deepspeed/runtime/zero/mics.py b/deepspeed/runtime/zero/mics.py
index f1b6b955239c..a1d34cc70c3e 100755
--- a/deepspeed/runtime/zero/mics.py
+++ b/deepspeed/runtime/zero/mics.py
@@ -408,7 +408,7 @@ def allreduce_mics_shard_grads(self, params, partitioned_grads_buffers: List[Ten
         """
         """
         # TODO: improve the condition check
-        if not self.is_gradient_accumulation_boundary or \
+        if not self.is_gradient_accumulation_boundary() or \
             len(partitioned_grads_buffers) == 0:
             return
 
diff --git a/deepspeed/runtime/zero/stage3.py b/deepspeed/runtime/zero/stage3.py
index 93cead30f090..f60f401f6fa6 100644
--- a/deepspeed/runtime/zero/stage3.py
+++ b/deepspeed/runtime/zero/stage3.py
@@ -375,6 +375,8 @@ def _enforce_optimizer_offload():
         self.postscale_gradients = postscale_gradients
         self.gradient_accumulation_steps = gradient_accumulation_steps
         self.micro_step_id = 0
+        # ds_ids reduced since the last step(); used to finalize only active params in unmanaged offload mode.
+        self._offload_boundary_param_ids = set()
         self.reduce_bucket_size = int(reduce_bucket_size)
 
         if self.all2all_process_group is not None:
@@ -439,8 +441,6 @@ def _enforce_optimizer_offload():
         if self.swap_optimizer:
             self._configure_tensor_swapping(offload_optimizer_config, aio_config)
 
-        self.is_gradient_accumulation_boundary: bool = True
-
         # Toggled by DeepSpeedEngine.coalesce_grad_reduction().
         self._coalesce_grad_reduction = False
 
@@ -1353,10 +1353,52 @@ def independent_gradient_partition_epilogue(self):
         self._epilogue_ran_this_backward = True
 
     def finalize_gradient_accumulation_boundary(self):
-        # Unmanaged mode: grad partitions already accumulate across backwards via __param_id_to_grad_partition; nothing to finalize for non-offload.
-        assert not self.offload_optimizer and not self.offload_param, \
-            "unmanaged gradient accumulation does not support ZeRO offload"
-        self.is_gradient_accumulation_boundary = True
+        # Unmanaged mode: partitions already accumulate in __param_id_to_grad_partition; offload still needs deferred boundary copy.
+        # Mirror engine boundary for any managed-style readers during step(); finalize itself does not branch on it.
+        self.set_gradient_accumulation_boundary(True)
+        if self.offload_optimizer:
+            self._finalize_offload_gradient_accumulation()
+
+    def _offload_grad_partition_at_boundary(self, param, grad_buffer, offload_fp32_gradients, offload_fp32_offsets):
+        # Boundary-only: record grad norm and copy/swap into optimizer FP32 or NVMe buffers.
+        i, dest_offset, _ = self.grad_position[self.get_param_id(param)]
+        self.norm_for_param_grads[self.get_param_id(param)] = self._constant_buffered_norm2(grad_buffer)
+
+        if self._swappable_optimizer_subgroup(i):
+            if i not in offload_fp32_gradients.keys():
+                offload_fp32_gradients[i] = []
+                offload_fp32_offsets[i] = []
+
+            offload_fp32_gradients[i].append(grad_buffer.to(dtype=self.master_weights_and_grads_dtype))
+            offload_fp32_offsets[i].append(dest_offset)
+        else:
+            fp32_grad_tensor = self.fp32_partitioned_groups_flat[i].grad.narrow(0, dest_offset, grad_buffer.numel())
+            fp32_grad_tensor.copy_(grad_buffer.to(dtype=self.master_weights_and_grads_dtype), non_blocking=True)
+
+    def _swap_out_offload_fp32_gradients(self, offload_fp32_gradients, offload_fp32_offsets):
+        if not (self.offload_optimizer and self.swap_optimizer):
+            return
+        for i in offload_fp32_gradients.keys():
+            self.optimizer_swapper.swap_out_gradients(parameter=self.fp32_partitioned_groups_flat[i],
+                                                      gradient_offsets=offload_fp32_offsets[i],
+                                                      gradient_tensors=offload_fp32_gradients[i])
+
+    def _finalize_offload_gradient_accumulation(self):
+        # Deferred boundary work for params reduced this window (matches managed offload; skips inactive params).
+        offload_fp32_gradients = {}
+        offload_fp32_offsets = {}
+        for param_group in self.fp16_groups:
+            for param in param_group:
+                if param.ds_id not in self._offload_boundary_param_ids:
+                    continue
+                if param.ds_id not in self.__param_id_to_grad_partition:
+                    continue
+                grad_buffer = self.__param_id_to_grad_partition[param.ds_id]
+                if not get_accelerator().on_accelerator(grad_buffer):
+                    grad_buffer = grad_buffer.to(get_accelerator().current_device_name(), non_blocking=True)
+                self._offload_grad_partition_at_boundary(param, grad_buffer, offload_fp32_gradients,
+                                                         offload_fp32_offsets)
+        self._swap_out_offload_fp32_gradients(offload_fp32_gradients, offload_fp32_offsets)
 
     def overlapping_partition_gradients_reduce_epilogue(self):
         self.independent_gradient_partition_epilogue()
@@ -1822,6 +1864,9 @@ def partition_grads(self, params_to_release: List[Parameter], grad_partitions: L
                 param.grad = None
                 continue
 
+            # Record active param so unmanaged offload finalize skips params unused this window.
+            self._offload_boundary_param_ids.add(param.ds_id)
+
             # move or accumulate gradient partition to target buffer
             grad_buffer = self.__param_id_to_grad_partition[param.ds_id].narrow(0, 0, grad_partition.numel())
             buffers.append(grad_buffer)
@@ -1844,23 +1889,9 @@ def partition_grads(self, params_to_release: List[Parameter], grad_partitions: L
 
             # offload the gradient partition if applicable
             if self.offload_optimizer:
-                i, dest_offset, _ = self.grad_position[self.get_param_id(param)]
-
-                if self.is_gradient_accumulation_boundary:
-                    self.norm_for_param_grads[self.get_param_id(param)] = self._constant_buffered_norm2(grad_buffer)
-
-                    if self._swappable_optimizer_subgroup(i):
-                        if i not in offload_fp32_gradients.keys():
-                            offload_fp32_gradients[i] = []
-                            offload_fp32_offsets[i] = []
-
-                        offload_fp32_gradients[i].append(grad_buffer.to(dtype=self.master_weights_and_grads_dtype))
-                        offload_fp32_offsets[i].append(dest_offset)
-                    else:
-                        fp32_grad_tensor = self.fp32_partitioned_groups_flat[i].grad.narrow(
-                            0, dest_offset, grad_buffer.numel())
-                        fp32_grad_tensor.copy_(grad_buffer.to(dtype=self.master_weights_and_grads_dtype),
-                                               non_blocking=True)
+                if self.is_gradient_accumulation_boundary():
+                    self._offload_grad_partition_at_boundary(param, grad_buffer, offload_fp32_gradients,
+                                                             offload_fp32_offsets)
 
             # free the gradient
             if not get_accelerator().is_synchronized_device():
@@ -1868,11 +1899,7 @@ def partition_grads(self, params_to_release: List[Parameter], grad_partitions: L
                     param.grad.record_stream(get_accelerator().current_stream())
             param.grad = None
 
-        if self.offload_optimizer and self.swap_optimizer:
-            for i in offload_fp32_gradients.keys():
-                self.optimizer_swapper.swap_out_gradients(parameter=self.fp32_partitioned_groups_flat[i],
-                                                          gradient_offsets=offload_fp32_offsets[i],
-                                                          gradient_tensors=offload_fp32_gradients[i])
+        self._swap_out_offload_fp32_gradients(offload_fp32_gradients, offload_fp32_offsets)
         return buffers
 
     def _partitioned_buffers_all_gather(self, params: List[Parameter], buffers_to_allgather: List[Tensor],
@@ -2306,6 +2333,7 @@ def reset_cpu_buffers(self):
 
     def _pre_step(self):
         self.micro_step_id = 0
+        self._offload_boundary_param_ids = set()
         # Also reset the epilogue flag so the next iteration starts fresh.
         # Without this, the flag from the last backward before step() would cause
         # an increment in the next forward(), which is wrong.
diff --git a/deepspeed/runtime/zero/stage_1_and_2.py b/deepspeed/runtime/zero/stage_1_and_2.py
index 85dd6ffb46b5..960854ad248c 100755
--- a/deepspeed/runtime/zero/stage_1_and_2.py
+++ b/deepspeed/runtime/zero/stage_1_and_2.py
@@ -252,8 +252,6 @@ def __init__(self,
         self.real_dp_process_group = [dp_process_group for i in range(len(self.optimizer.param_groups))]
         self.partition_count = [dp_size for i in range(len(self.optimizer.param_groups))]
 
-        self.is_gradient_accumulation_boundary = True
-
         # Toggled by DeepSpeedEngine.coalesce_grad_reduction().
         self._coalesce_grad_reduction = False
 
@@ -579,6 +577,8 @@ def _enforce_cpu_offload():
             self.norm_for_param_grads = {}
             self.local_overflow = False
             self.grad_position = {}
+            # Param ids reduced since the last step(); used to finalize only active params in unmanaged mode.
+            self._offload_accumulated_param_ids = set()
             self.temp_grad_buffer_for_cpu_offload = torch.zeros(largest_param_numel,
                                                                 device=self.device,
                                                                 dtype=self.dtype)
@@ -921,7 +921,7 @@ def independent_gradient_partition_epilogue(self):
                                                         dtype=self.gradient_accumulation_dtype)
                     for accumulated_grad, new_avg_grad in zip(self.all_grad_tensors[i], avg_new):
                         accumulated_grad.add_(new_avg_grad)
-                if self.is_gradient_accumulation_boundary:
+                if self.is_gradient_accumulation_boundary():
                     self.averaged_gradients[i] = self.get_flat_partition(
                         self.params_in_partition[i],
                         self.first_offset[i],
@@ -948,9 +948,12 @@ def independent_gradient_partition_epilogue(self):
         see_memory_usage("End ipg_epilogue")
 
     def finalize_gradient_accumulation_boundary(self):
-        # Unmanaged mode: grads were reduced/accumulated into all_grad_tensors each backward; finalize averaged_gradients for step().
-        assert not self.cpu_offload, "unmanaged gradient accumulation does not support ZeRO optimizer state offload"
-        self.is_gradient_accumulation_boundary = True
+        # Unmanaged mode: grads accumulated each backward; finalize for step() (averaged_gradients or offload fp32 copy).
+        # Mirror engine boundary for any managed-style readers during step(); finalize itself does not branch on it.
+        self.set_gradient_accumulation_boundary(True)
+        if self.cpu_offload:
+            self._finalize_cpu_offload_gradient_accumulation()
+            return
         for i, _ in enumerate(self.bit16_groups):
             self.averaged_gradients[i] = self.get_flat_partition(self.params_in_partition[i],
                                                                  self.first_offset[i],
@@ -961,6 +964,36 @@ def finalize_gradient_accumulation_boundary(self):
                                                                  return_tensor_list=True)
             self.all_grad_tensors[i] = None
 
+    def _finalize_cpu_offload_gradient_accumulation(self):
+        # Deferred boundary work for params reduced this window (matches managed offload; skips inactive params).
+        for group in self.params_in_partition:
+            for param in group:
+                if not param.requires_grad:
+                    continue
+                if self.get_param_id(param) not in self._offload_accumulated_param_ids:
+                    continue
+                self._restore_cpu_offload_grad_to_gpu(param)
+                self.set_norm_for_param_grad_in_gpu(param)
+                self.update_offload_overflow_tracker_for_param_grad(param)
+                self.async_inplace_copy_grad_to_fp32_buffer_from_gpu(param)
+
+    def _restore_cpu_offload_grad_to_gpu(self, param):
+        # Last non-boundary epilogue cleared param.grad; reload accumulated CPU grads for boundary helpers.
+        param_id = self.get_param_id(param)
+        [_, source_offset, dest_offset, num_elements] = self.grad_position[param_id]
+        dest_buffer = self.temp_grad_buffer_for_gpu_offload.view(-1).narrow(0, 0, param.numel())
+        if not self.low_precision_master_weights_and_grads:
+            dest_buffer.copy_(self.accumulated_grads_in_cpu[param_id].view(-1), non_blocking=True)
+        else:
+            dest_buffer.narrow(0, source_offset, num_elements).copy_(self.accumulated_grads_in_cpu[param_id].view(-1),
+                                                                     non_blocking=True)
+        # Clone so the shared temp buffer can be reused for the next parameter.
+        restored = dest_buffer.view_as(param).clone()
+        if self.use_grad_accum_attribute:
+            param.grad_accum = restored
+        else:
+            param.grad = restored
+
     def clear_backward_seen_flag(self):
         """Clear the backward seen flag and do deferred cleanup.
 
@@ -1593,10 +1626,12 @@ def copy_grads_in_partition(self, param):
             # CPU buffer) or more will follow (save to CPU buffer). Skipping only
             # the lone backward of a step preserves the existing fast path for
             # ga_steps=1 + single backward.
-            if self.micro_step_id > 0 or not self.is_gradient_accumulation_boundary:
+            if self.micro_step_id > 0 or not self.is_gradient_accumulation_boundary():
                 self.async_accumulate_grad_in_cpu_via_gpu(param)
+                # Record active param so unmanaged finalize skips params unused this window.
+                self._offload_accumulated_param_ids.add(self.get_param_id(param))
 
-            if self.is_gradient_accumulation_boundary:
+            if self.is_gradient_accumulation_boundary():
                 self.set_norm_for_param_grad_in_gpu(param)
 
                 self.update_offload_overflow_tracker_for_param_grad(param)
@@ -1703,7 +1738,7 @@ def process_gradients(self, param, i):
             self.reduce_ready_partitions_and_remove_grads(param, i)
 
     def reduce_ready_partitions_and_remove_grads(self, param, i):
-        if self.partition_gradients or self.is_gradient_accumulation_boundary or self.zenflow:
+        if self.partition_gradients or self.is_gradient_accumulation_boundary() or self.zenflow:
             self.reduce_independent_p_g_buckets_and_remove_grads(param, i)
 
     def zero_reduced_gradients(self, partition_id, i):
@@ -2220,6 +2255,8 @@ def step(self, closure=None):
         Not supporting closure.
         """
         self.micro_step_id = INITIAL_MICRO_STEP_ID
+        if self.cpu_offload:
+            self._offload_accumulated_param_ids = set()
 
         see_memory_usage("In step before checking overflow")
 
diff --git a/docs/_pages/config-json.md b/docs/_pages/config-json.md
index f09183f36978..8f91414abd8f 100755
--- a/docs/_pages/config-json.md
+++ b/docs/_pages/config-json.md
@@ -33,7 +33,7 @@ toc_label: "Contents"
 
 | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              | Default |
 | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------- |
-| Controls how gradient accumulation boundaries are managed. When `true`, DeepSpeed tracks micro-steps and applies the optimizer step only at the accumulation boundary, so `forward`/`backward`/`step` can be called symmetrically on every micro-batch. When `false`, micro-step tracking is disabled and the client is responsible for calling `step()` at the accumulation boundary; each `step()` finalizes the locally-accumulated gradients and applies an optimizer update. The `false` setting currently supports ZeRO stage 0/1/2/3 (and DDP); ZeRO offload (optimizer state and parameter) is planned but not yet available (rejected at initialization). It is also incompatible with pipeline parallelism, DeepCompile, and Apex AMP. ZeRO `overlap_comm` is supported only with ZeRO stage 2 (rejected for stage 0/1, where reduction is deferred to `step()`). | `true`  |
+| Controls how gradient accumulation boundaries are managed. When `true`, DeepSpeed tracks micro-steps and applies the optimizer step only at the accumulation boundary, so `forward`/`backward`/`step` can be called symmetrically on every micro-batch. When `false`, micro-step tracking is disabled and the client is responsible for calling `step()` at the accumulation boundary; each `step()` finalizes the locally-accumulated gradients and applies an optimizer update. The `false` setting supports ZeRO stage 0/1/2/3 (and DDP), including ZeRO optimizer-state and parameter offload (CPU/NVMe). It is incompatible with pipeline parallelism, DeepCompile, and Apex AMP. ZeRO `overlap_comm` is supported only with ZeRO stage 2 (rejected for stage 0/1, where reduction is deferred to `step()`). | `true`  |
 
 
 
diff --git a/docs/code-docs/source/training.rst b/docs/code-docs/source/training.rst
index 2cc4f80ff789..afa98e80c7d6 100644
--- a/docs/code-docs/source/training.rst
+++ b/docs/code-docs/source/training.rst
@@ -104,11 +104,13 @@ This is useful when ``backward`` and ``step`` must be decoupled and the number o
 client- or RPC-driven RL backends where a single optimizer step arrives as ``N`` ``backward()``
 calls followed by one ``step()``, with ``N`` unknown at configuration time.
 
-Unmanaged mode currently supports **ZeRO stage 0/1/2/3 and DDP**. For ZeRO stage 0/1 (and DDP),
-``backward()`` accumulates gradients locally and ``step()`` performs the gradient all-reduce followed
-by the optimizer update. For ZeRO stage 2/3, gradients are reduced/partitioned on every ``backward()``
-(as in managed mode) and ``step()`` finalizes the accumulated partition gradients before the optimizer
-update; ZeRO ``overlap_comm`` is supported in the stage-2 path.
+Unmanaged mode currently supports **ZeRO stage 0/1/2/3 and DDP**, including ZeRO optimizer-state
+and parameter offload (CPU/NVMe). For ZeRO stage 0/1 (and DDP), ``backward()`` accumulates gradients
+locally and ``step()`` performs the gradient all-reduce followed by the optimizer update. For ZeRO
+stage 2/3, gradients are reduced/partitioned on every ``backward()`` (as in managed mode) and
+``step()`` finalizes the accumulated partition gradients (including deferred offload norm/FP32 or
+NVMe copy when offload is enabled) before the optimizer update; ZeRO ``overlap_comm`` is supported
+in the stage-2 path.
 
 .. note::
    By default ``backward()`` scales the loss and gradients by the configured
@@ -120,12 +122,10 @@ update; ZeRO ``overlap_comm`` is supported in the stage-2 path.
    applies and no manual averaging is needed.)
 
 .. note::
-   Unmanaged mode is being added incrementally. ZeRO stage 0/1/2/3 (and DDP) is supported today;
-   ZeRO offload (optimizer state and parameter) is planned but **not yet available** -- enabling it
-   with ``managed_gradient_accumulation=false`` raises an ``AssertionError`` at initialization.
-   Unmanaged mode is likewise incompatible with pipeline parallelism, DeepCompile, and Apex AMP,
-   which are also rejected at initialization. ZeRO ``overlap_comm`` is supported only with ZeRO
-   stage 2 (rejected for stage 0/1, where reduction is deferred to ``step()``).
+   Unmanaged mode supports ZeRO stage 0/1/2/3 (and DDP), including ZeRO optimizer-state and parameter
+   offload (CPU/NVMe). It is incompatible with pipeline parallelism, DeepCompile, and Apex AMP, which
+   are rejected at initialization. ZeRO ``overlap_comm`` is supported only with ZeRO stage 2
+   (rejected for stage 0/1, where reduction is deferred to ``step()``).
 
 .. autofunction:: deepspeed.DeepSpeedEngine.set_gradient_accumulation_boundary
 
diff --git a/tests/unit/v1/zero/test_zero_user_backward.py b/tests/unit/v1/zero/test_zero_user_backward.py
index 54059eb2720a..1b75675491d7 100644
--- a/tests/unit/v1/zero/test_zero_user_backward.py
+++ b/tests/unit/v1/zero/test_zero_user_backward.py
@@ -14,6 +14,8 @@
 from deepspeed.accelerator import get_accelerator
 from deepspeed.utils import safe_get_full_grad
 from deepspeed.runtime.zero.partition_parameters import ZeroParamStatus
+from deepspeed.ops.aio import AsyncIOBuilder
+from deepspeed.runtime.zero.offload_config import OffloadDeviceEnum
 
 
 class SimpleNonScalarModel(torch.nn.Module):
@@ -2042,29 +2044,9 @@ def test_set_gradient_accumulation_boundary_rejected(self, zero_stage):
 
 
 class TestUnmanagedGradientAccumulationValidation(DistributedTest):
-    """Unmanaged mode rejects ZeRO offload (until follow-up PR)."""
+    """Unmanaged mode accepts disabled offload template blocks (device=none)."""
     world_size = 1
 
-    def test_unmanaged_rejects_zero_offload(self):
-        hidden_dim = 4
-        initialize_distributed()
-        torch.manual_seed(42)
-        model = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
-        config = build_managed_gas_config(2, gradient_accumulation_steps=1, managed_gradient_accumulation=False)
-        config["zero_optimization"]["offload_optimizer"] = {"device": "cpu"}
-        with pytest.raises(AssertionError, match="not supported with ZeRO optimizer state offload"):
-            deepspeed.initialize(config=config, model=model, model_parameters=model.parameters())
-
-    def test_unmanaged_rejects_param_offload(self):
-        hidden_dim = 4
-        initialize_distributed()
-        torch.manual_seed(42)
-        model = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
-        config = build_managed_gas_config(3, gradient_accumulation_steps=1, managed_gradient_accumulation=False)
-        config["zero_optimization"]["offload_param"] = {"device": "cpu"}
-        with pytest.raises(AssertionError, match="not supported with ZeRO parameter offload"):
-            deepspeed.initialize(config=config, model=model, model_parameters=model.parameters())
-
     def test_unmanaged_accepts_disabled_offload_blocks(self):
         # A disabled offload block (device="none") is not offload, so init must succeed.
         hidden_dim = 4
@@ -2079,19 +2061,230 @@ def test_unmanaged_accepts_disabled_offload_blocks(self):
         engine.destroy()
 
 
-class TestUnmanagedGradientAccumulationOffloadValidation(DistributedTest):
-    """Unmanaged mode does not support ZeRO optimizer offload (stage 1: before the stage-2/3 guard)."""
-    world_size = 1
+def _build_unmanaged_offload_config(zero_stage,
+                                    gradient_accumulation_steps,
+                                    managed,
+                                    offload_param=False,
+                                    offload_device="cpu",
+                                    nvme_path=None):
+    config = build_managed_gas_config(zero_stage, gradient_accumulation_steps, managed_gradient_accumulation=managed)
+    offload_opt = {"device": offload_device}
+    if offload_device == OffloadDeviceEnum.nvme:
+        assert nvme_path is not None, "nvme_path is required for NVMe offload"
+        offload_opt["nvme_path"] = str(nvme_path)
+    config["zero_optimization"]["offload_optimizer"] = offload_opt
+    if offload_param:
+        offload_p = {"device": offload_device}
+        if offload_device == OffloadDeviceEnum.nvme:
+            offload_p["nvme_path"] = str(nvme_path)
+        config["zero_optimization"]["offload_param"] = offload_p
+    if offload_device == OffloadDeviceEnum.nvme:
+        # Small sub_group_size so each large param is its own swappable subgroup.
+        config["zero_optimization"]["sub_group_size"] = 100
+        config["aio"] = {"block_size": 1048576}
+    return config
+
+
+def _run_unmanaged_vs_managed_offload(zero_stage,
+                                      gradient_accumulation_steps,
+                                      num_cycles,
+                                      offload_param=False,
+                                      offload_device="cpu",
+                                      nvme_path=None,
+                                      hidden_dim=4,
+                                      label=""):
+    device, _, _ = initialize_distributed()
+
+    def make_config(managed):
+        return _build_unmanaged_offload_config(zero_stage,
+                                               gradient_accumulation_steps,
+                                               managed=managed,
+                                               offload_param=offload_param,
+                                               offload_device=offload_device,
+                                               nvme_path=nvme_path)
+
+    torch.manual_seed(42)
+    if zero_stage == 3 and offload_device == OffloadDeviceEnum.nvme:
+        with deepspeed.zero.Init(config_dict_or_path=make_config(True)):
+            model_managed = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
+    else:
+        model_managed = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
+    managed_engine, _, _, _ = deepspeed.initialize(config=make_config(True),
+                                                   model=model_managed,
+                                                   model_parameters=model_managed.parameters())
+
+    torch.manual_seed(42)
+    if zero_stage == 3 and offload_device == OffloadDeviceEnum.nvme:
+        with deepspeed.zero.Init(config_dict_or_path=make_config(False)):
+            model_unmanaged = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
+    else:
+        model_unmanaged = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
+    unmanaged_engine, _, _, _ = deepspeed.initialize(config=make_config(False),
+                                                     model=model_unmanaged,
+                                                     model_parameters=model_unmanaged.parameters())
+
+    total_samples = num_cycles * gradient_accumulation_steps
+    batches = list(
+        random_dataloader(model=managed_engine,
+                          total_samples=total_samples,
+                          hidden_dim=hidden_dim,
+                          device=device,
+                          dtype=torch.float32))
+
+    for batch in batches:
+        loss = managed_engine(batch[0], batch[1])
+        managed_engine.backward(loss)
+        managed_engine.step()
+
+    for cycle in range(num_cycles):
+        for micro in range(gradient_accumulation_steps):
+            batch = batches[cycle * gradient_accumulation_steps + micro]
+            loss = unmanaged_engine(batch[0], batch[1])
+            unmanaged_engine.backward(loss)
+        unmanaged_engine.step()
+
+    managed_params = collect_deepspeed_parameters(managed_engine, zero_stage)
+    unmanaged_params = collect_deepspeed_parameters(unmanaged_engine, zero_stage)
+    compare_parameters(managed_params, unmanaged_params, label)
+
+    managed_engine.destroy()
+    unmanaged_engine.destroy()
+
+
+@pytest.mark.parametrize("zero_stage", [1, 2, 3])
+class TestUnmanagedGradientAccumulationOffload(DistributedTest):
+    """Unmanaged mode with ZeRO optimizer offload matches managed offload."""
+    world_size = 2
+
+    def test_unmanaged_matches_managed_optimizer_offload(self, zero_stage):
+        _run_unmanaged_vs_managed_offload(zero_stage,
+                                          gradient_accumulation_steps=4,
+                                          num_cycles=3,
+                                          label=f"unmanaged vs managed optimizer offload (stage {zero_stage})")
+
+
+class TestUnmanagedGradientAccumulationParamOffload(DistributedTest):
+    """Unmanaged ZeRO-3 with parameter + optimizer offload matches managed mode."""
+    world_size = 2
+
+    def test_unmanaged_matches_managed_param_offload(self):
+        _run_unmanaged_vs_managed_offload(zero_stage=3,
+                                          gradient_accumulation_steps=4,
+                                          num_cycles=2,
+                                          offload_param=True,
+                                          label="unmanaged vs managed param+optimizer offload (stage 3)")
+
+
+@pytest.mark.sequential
+class TestUnmanagedGradientAccumulationNvmeOffload(DistributedTest):
+    """Unmanaged ZeRO-3 NVMe offload matches managed NVMe offload."""
+    world_size = 2
+
+    def _skip_if_nvme_unsupported(self):
+        if not deepspeed.ops.__compatible_ops__[AsyncIOBuilder.NAME]:
+            pytest.skip("Skip tests since async-io is not compatible")
+
+    def test_unmanaged_matches_managed_optimizer_nvme_offload(self, tmpdir):
+        self._skip_if_nvme_unsupported()
+        # Large enough that partitioned params exceed MIN_AIO_BYTES and exercise swap_out_gradients.
+        _run_unmanaged_vs_managed_offload(zero_stage=3,
+                                          gradient_accumulation_steps=2,
+                                          num_cycles=2,
+                                          offload_device=OffloadDeviceEnum.nvme,
+                                          nvme_path=tmpdir,
+                                          hidden_dim=1024,
+                                          label="unmanaged vs managed optimizer NVMe offload (stage 3)")
+
+    def test_unmanaged_matches_managed_param_nvme_offload(self, tmpdir):
+        self._skip_if_nvme_unsupported()
+        _run_unmanaged_vs_managed_offload(zero_stage=3,
+                                          gradient_accumulation_steps=2,
+                                          num_cycles=2,
+                                          offload_param=True,
+                                          offload_device=OffloadDeviceEnum.nvme,
+                                          nvme_path=tmpdir,
+                                          hidden_dim=1024,
+                                          label="unmanaged vs managed param+optimizer NVMe offload (stage 3)")
+
+
+class _TwoHeadModel(torch.nn.Module):
+    """Two independent heads; only one is exercised per window so the other receives no gradient."""
+
+    def __init__(self, hidden_dim):
+        super().__init__()
+        self.head_a = torch.nn.Linear(hidden_dim, hidden_dim, bias=False)
+        self.head_b = torch.nn.Linear(hidden_dim, hidden_dim, bias=False)
+        self.cross_entropy_loss = torch.nn.CrossEntropyLoss()
+        self.use_a = True
+
+    def forward(self, x, y):
+        out = self.head_a(x) if self.use_a else self.head_b(x)
+        return self.cross_entropy_loss(out, y)
 
-    def test_unmanaged_rejects_offload(self):
+
+@pytest.mark.parametrize("zero_stage", [1, 2, 3])
+class TestUnmanagedGradientAccumulationOffloadInactiveParams(DistributedTest):
+    """Unmanaged optimizer offload must skip params that receive no gradient in a window (matches managed)."""
+    world_size = 2
+
+    def test_unmanaged_matches_managed_inactive_params(self, zero_stage):
         hidden_dim = 4
-        initialize_distributed()
-        torch.manual_seed(42)
-        model = SimpleModel(hidden_dim=hidden_dim, nlayers=2)
-        config = build_managed_gas_config(1, gradient_accumulation_steps=1, managed_gradient_accumulation=False)
-        config["zero_optimization"]["offload_optimizer"] = {"device": "cpu"}
-        with pytest.raises(AssertionError, match="not supported with ZeRO optimizer state offload"):
-            deepspeed.initialize(config=config, model=model, model_parameters=model.parameters())
+        gradient_accumulation_steps = 2
+        # One head is exclusively active per window, so the other head is inactive that whole window.
+        window_use_a = [True, False, True]
+
+        device, _, _ = initialize_distributed()
+
+        def offload_config(managed):
+            config = _build_unmanaged_offload_config(zero_stage, gradient_accumulation_steps, managed=managed)
+            config["zero_optimization"]["ignore_unused_parameters"] = True
+            return config
+
+        torch.manual_seed(123)
+        model_managed = _TwoHeadModel(hidden_dim)
+        managed_engine, _, _, _ = deepspeed.initialize(config=offload_config(True),
+                                                       model=model_managed,
+                                                       model_parameters=model_managed.parameters())
+
+        torch.manual_seed(123)
+        model_unmanaged = _TwoHeadModel(hidden_dim)
+        unmanaged_engine, _, _, _ = deepspeed.initialize(config=offload_config(False),
+                                                         model=model_unmanaged,
+                                                         model_parameters=model_unmanaged.parameters())
+
+        total_samples = len(window_use_a) * gradient_accumulation_steps
+        batches = list(
+            random_dataloader(model=managed_engine,
+                              total_samples=total_samples,
+                              hidden_dim=hidden_dim,
+                              device=device,
+                              dtype=torch.float32))
+
+        # Managed: symmetric forward/backward/step; optimizer applies on each GAS boundary.
+        for w, use_a in enumerate(window_use_a):
+            managed_engine.module.use_a = use_a
+            for micro in range(gradient_accumulation_steps):
+                batch = batches[w * gradient_accumulation_steps + micro]
+                loss = managed_engine(batch[0], batch[1])
+                managed_engine.backward(loss)
+                managed_engine.step()
+
+        # Unmanaged: N backwards then one step() per window.
+        for w, use_a in enumerate(window_use_a):
+            unmanaged_engine.module.use_a = use_a
+            for micro in range(gradient_accumulation_steps):
+                batch = batches[w * gradient_accumulation_steps + micro]
+                loss = unmanaged_engine(batch[0], batch[1])
+                unmanaged_engine.backward(loss)
+            unmanaged_engine.step()
+
+        managed_params = collect_deepspeed_parameters(managed_engine, zero_stage)
+        unmanaged_params = collect_deepspeed_parameters(unmanaged_engine, zero_stage)
+        compare_parameters(managed_params, unmanaged_params,
+                           f"unmanaged vs managed offload inactive params (stage {zero_stage})")
+
+        managed_engine.destroy()
+        unmanaged_engine.destroy()
 
 
 @pytest.mark.parametrize("zero_stage", [0, 1])