diff --git a/docs/api/pytorch.rst b/docs/api/pytorch.rst
index 5fac0a89a6..84d850eebc 100644
--- a/docs/api/pytorch.rst
+++ b/docs/api/pytorch.rst
@@ -40,6 +40,8 @@ PyTorch
 
 .. autoapiclass:: transformer_engine.pytorch.autocast(enabled=True, calibrating=False, recipe=None, amax_reduction_group=None)
 
+.. autoapifunction:: transformer_engine.pytorch.quantization_backward_scope
+
 .. autoapifunction:: transformer_engine.pytorch.quantized_model_init
 
 .. autoapifunction:: transformer_engine.pytorch.checkpoint
diff --git a/docs/features/low_precision_training/fp8_delayed_scaling/fp8_delayed_scaling.rst b/docs/features/low_precision_training/fp8_delayed_scaling/fp8_delayed_scaling.rst
index 9d05305eda..30665018ff 100644
--- a/docs/features/low_precision_training/fp8_delayed_scaling/fp8_delayed_scaling.rst
+++ b/docs/features/low_precision_training/fp8_delayed_scaling/fp8_delayed_scaling.rst
@@ -146,6 +146,19 @@ However, amax reduction works slightly differently in different frameworks.
         skipped - if no rank executes a module, its history is not rotated and scale 
         remains unchanged.
 
+      The gradient amaxes are reduced separately, once at the end of every ``backward()``
+      call, on each rank where at least one quantized module ran backward. When a
+      training step consists of several ``backward()`` calls (e.g. a 1F1B pipeline
+      schedule), wrap them in ``quantization_backward_scope`` so the update runs once
+      per step. The scope also
+      runs the update on ranks where no quantized module ran backward, so, like
+      ``autocast``, it must be entered and exited on all ranks:
+
+      .. literalinclude:: pytorch_delayed_scaling_distributed_example.py
+         :language: python
+         :start-after: # START_BACKWARD_SCOPE_EXAMPLE
+         :end-before: # END_BACKWARD_SCOPE_EXAMPLE
+
 
    .. tab:: JAX
 
diff --git a/docs/features/low_precision_training/fp8_delayed_scaling/pytorch_delayed_scaling_distributed_example.py b/docs/features/low_precision_training/fp8_delayed_scaling/pytorch_delayed_scaling_distributed_example.py
index 863b71e8c6..230e7fb7f0 100644
--- a/docs/features/low_precision_training/fp8_delayed_scaling/pytorch_delayed_scaling_distributed_example.py
+++ b/docs/features/low_precision_training/fp8_delayed_scaling/pytorch_delayed_scaling_distributed_example.py
@@ -16,3 +16,9 @@
     output = model(inp)
 
 # END_AMAX_REDUCTION_EXAMPLE
+
+# START_BACKWARD_SCOPE_EXAMPLE
+with te.quantization_backward_scope():
+    for loss in microbatch_losses:
+        loss.backward()
+# END_BACKWARD_SCOPE_EXAMPLE
diff --git a/tests/pytorch/distributed/run_numerics.py b/tests/pytorch/distributed/run_numerics.py
index fe02f990b4..5d8529e909 100644
--- a/tests/pytorch/distributed/run_numerics.py
+++ b/tests/pytorch/distributed/run_numerics.py
@@ -28,6 +28,7 @@
 from transformer_engine.pytorch import Float8CurrentScalingQuantizer, NVFP4Quantizer
 from transformer_engine.pytorch.constants import NVFP4_BLOCK_SCALING_SIZE
 from transformer_engine.pytorch.distributed import gather_along_first_dim
+from transformer_engine.pytorch.quantization import FP8GlobalStateManager
 from run_layer_with_overlap import _compare_tensors
 
 SEQ_LEN, BATCH_SIZE = 16, 16
@@ -132,6 +133,7 @@ def main(argv=None, namespace=None):
         test_layernorm_linear,
         test_layernorm_mlp,
         test_transformer_layer,
+        test_backward_update_with_skipped_ranks,
     ]
 
     for test in test_dict:
@@ -1139,5 +1141,74 @@ def test_transformer_layer():
             _test_transformer_layer_parallel(sequence_parallel, **kwargs)
 
 
+############################################
+#      Delayed-scaling backward update     #
+############################################
+
+
+def _assert_bwd_state_matches_across_ranks(model):
+    for module in model:
+        state = module.fp8_meta["scaling_bwd"]
+        for t in (state.amax_history, state.scale):
+            gathered = [torch.empty_like(t) for _ in range(WORLD_SIZE)]
+            dist.all_gather(gathered, t)
+            for other in gathered[1:]:
+                assert torch.equal(other, gathered[0]), f"{gathered[0]} vs {other}"
+
+
+def _backward_update_step_skipped_module(model, recipe):
+    """Odd ranks feed the first module an empty batch and drop its output."""
+    rows = BATCH_SIZE if WORLD_RANK % 2 == 0 else 0
+    x_a = torch.randn(rows, HIDDEN_SIZE, device="cuda", requires_grad=True)
+    x_b = torch.randn(BATCH_SIZE, HIDDEN_SIZE, device="cuda", requires_grad=True)
+    with te.autocast(enabled=True, recipe=recipe):
+        y_a = model[0](x_a)
+        y_b = model[1](x_b)
+    loss = y_b.float().sum()
+    if y_a.numel() > 0:
+        loss = loss + y_a.float().sum()
+    loss.backward()
+
+
+def _backward_update_step_no_backward_in_scope(model, recipe):
+    """Odd ranks run no backward at all; the scope still triggers the update."""
+    x = torch.randn(BATCH_SIZE, HIDDEN_SIZE, device="cuda", requires_grad=True)
+    with te.quantization_backward_scope():
+        with te.autocast(enabled=True, recipe=recipe):
+            y = model[1](model[0](x))
+        if WORLD_RANK % 2 == 0:
+            y.float().sum().backward()
+
+
+@run_distributed_test()
+def _test_backward_update_with_skipped_ranks(step_fn):
+    # Drop amax buffers registered by earlier tests so only this model is reduced.
+    FP8GlobalStateManager.reset()
+    model = nn.ModuleList([te.Linear(HIDDEN_SIZE, HIDDEN_SIZE, bias=True) for _ in range(2)]).cuda()
+    recipe = DelayedScaling(reduce_amax=True)
+    qstate = FP8GlobalStateManager.quantization_state
+    for _ in range(3):
+        model.zero_grad(set_to_none=True)
+        step_fn(model, recipe)
+        assert not qstate.pending_backward_quantization_update
+        assert qstate.backward_quantization_update_callback_task_id is None
+        _assert_bwd_state_matches_across_ranks(model)
+    # Ranks that skipped backward must have received the other ranks' amaxes.
+    for module in model:
+        assert module.fp8_meta["scaling_bwd"].amax_history.abs().sum() > 0
+
+
+def test_backward_update_with_skipped_ranks():
+    """Every rank must join the amax reduction even if it skipped backward."""
+    if QUANTIZATION != "fp8":
+        return
+    for step_fn in (
+        _backward_update_step_skipped_module,
+        _backward_update_step_no_backward_in_scope,
+    ):
+        _test_backward_update_with_skipped_ranks(step_fn)
+    FP8GlobalStateManager.reset()
+
+
 if __name__ == "__main__":
     sys.exit(main())
diff --git a/tests/pytorch/test_backward_override.py b/tests/pytorch/test_backward_override.py
index c0acf2e6b3..3dc308ed72 100644
--- a/tests/pytorch/test_backward_override.py
+++ b/tests/pytorch/test_backward_override.py
@@ -408,7 +408,7 @@ def _snapshot_layout_invariants(
 
 def _snapshot_backward_ctx_state(
     output: torch.Tensor,
-) -> tuple[str, bool, object, bool]:
+) -> tuple[str, bool, object]:
     if output.grad_fn is None:
         raise RuntimeError("Output tensor has no grad_fn; cannot inspect backward context state.")
     # ``Linear`` packs backward state into ``grad_fn.backward_objects``
@@ -419,7 +419,6 @@ def _snapshot_backward_ctx_state(
         "backward_override",
         "fp8",
         "grad_output_quantizer",
-        "reduce_and_update_bwd_fp8_tensors",
     )
     missing_attrs = [attr for attr in required_attrs if not hasattr(state_holder, attr)]
     if missing_attrs:
@@ -430,7 +429,6 @@ def _snapshot_backward_ctx_state(
         getattr(state_holder, "backward_override"),
         bool(getattr(state_holder, "fp8")),
         getattr(state_holder, "grad_output_quantizer"),
-        bool(getattr(state_holder, "reduce_and_update_bwd_fp8_tensors")),
     )
 
 
@@ -816,7 +814,6 @@ def _run_grouped_linear_single_step_with_ctx_state(
         required_attrs = (
             "backward_override",
             "fp8",
-            "reduce_and_update_bwd_fp8_tensors",
         )
         missing_attrs = [attr for attr in required_attrs if not hasattr(y.grad_fn, attr)]
         if missing_attrs:
@@ -827,7 +824,6 @@ def _run_grouped_linear_single_step_with_ctx_state(
         ctx_state = (
             getattr(y.grad_fn, "backward_override"),
             bool(getattr(y.grad_fn, "fp8")),
-            bool(getattr(y.grad_fn, "reduce_and_update_bwd_fp8_tensors")),
         )
     y.backward(dy)
     assert x_run.grad is not None
@@ -1449,37 +1445,22 @@ def test_linear_like_runtime_backward_override_switch_updates_ctx(
     skip_unsupported_backward_override(module_type, mode_recipe, backward_override)
 
     *_, default_ctx = _run_single_step_with_ctx_state(module, x, dy, default_recipe)
-    (
-        default_mode,
-        default_fp8,
-        default_grad_output_quantizer,
-        default_reduce_and_update,
-    ) = default_ctx
+    default_mode, default_fp8, default_grad_output_quantizer = default_ctx
     assert default_mode is None
     assert default_fp8
     assert default_grad_output_quantizer is not None
-    assert default_reduce_and_update
 
     *_, switched_ctx = _run_single_step_with_ctx_state(module, x, dy, mode_recipe)
-    switched_mode, switched_fp8, switched_grad_output_quantizer, switched_reduce_and_update = (
-        switched_ctx
-    )
+    switched_mode, switched_fp8, switched_grad_output_quantizer = switched_ctx
     assert switched_mode == backward_override
     assert not switched_fp8
     assert switched_grad_output_quantizer is None
-    assert not switched_reduce_and_update
 
     *_, default_ctx_after = _run_single_step_with_ctx_state(module, x, dy, default_recipe)
-    (
-        default_mode_after,
-        default_fp8_after,
-        default_grad_output_quantizer_after,
-        default_reduce_and_update_after,
-    ) = default_ctx_after
+    default_mode_after, default_fp8_after, default_grad_output_quantizer_after = default_ctx_after
     assert default_mode_after is None
     assert default_fp8_after
     assert default_grad_output_quantizer_after is not None
-    assert default_reduce_and_update_after
 
 
 @pytest.mark.parametrize("recipe_name", _quantized_numerics_recipe_list)
@@ -1526,10 +1507,9 @@ def test_grouped_linear_runtime_backward_override_switch_updates_ctx(
         dy,
         default_recipe,
     )
-    default_mode, default_fp8, default_reduce_and_update = default_ctx
+    default_mode, default_fp8 = default_ctx
     assert default_mode is None
     assert default_fp8
-    assert default_reduce_and_update
 
     *_, switched_ctx = _run_grouped_linear_single_step_with_ctx_state(
         module,
@@ -1538,10 +1518,9 @@ def test_grouped_linear_runtime_backward_override_switch_updates_ctx(
         dy,
         mode_recipe,
     )
-    switched_mode, switched_fp8, switched_reduce_and_update = switched_ctx
+    switched_mode, switched_fp8 = switched_ctx
     assert switched_mode == backward_override
     assert not switched_fp8
-    assert not switched_reduce_and_update
 
     *_, default_ctx_after = _run_grouped_linear_single_step_with_ctx_state(
         module,
@@ -1550,10 +1529,9 @@ def test_grouped_linear_runtime_backward_override_switch_updates_ctx(
         dy,
         default_recipe,
     )
-    default_mode_after, default_fp8_after, default_reduce_and_update_after = default_ctx_after
+    default_mode_after, default_fp8_after = default_ctx_after
     assert default_mode_after is None
     assert default_fp8_after
-    assert default_reduce_and_update_after
 
 
 @pytest.mark.parametrize("recipe_name", _quantized_numerics_recipe_list)
diff --git a/tests/pytorch/test_numerics.py b/tests/pytorch/test_numerics.py
index e6a83d92bc..1a0bc6f7d9 100644
--- a/tests/pytorch/test_numerics.py
+++ b/tests/pytorch/test_numerics.py
@@ -13,6 +13,7 @@
 
 from transformer_engine.pytorch.quantization import (
     FP8GlobalStateManager,
+    quantization_backward_scope,
 )
 from transformer_engine.pytorch._extra_state import UNSAFE_PICKLE_EXTRA_STATE_ENV
 from transformer_engine.pytorch.utils import (
@@ -955,6 +956,219 @@ def body(value):
     assert _FP8_RECOMPUTE_KEY in fp8_layer.fp8_meta
 
 
+_UPDATE_TEST_HIDDEN = 128
+_UPDATE_TEST_BATCH = 32
+_UPDATE_TEST_STEPS = 3
+
+
+class _UpdateCounter:
+    """Count backward updates and check they run after every given module's backward.
+
+    A module that has run backward holds a nonzero grad amax in the current
+    history slot; the update rolls the history and clears that slot.
+    """
+
+    def __init__(self, *modules):
+        self.backward = 0
+        self._original = None
+        self._modules = [m for root in modules for m in root.modules() if hasattr(m, "fp8_meta")]
+
+    def _current_bwd_amax(self):
+        return [m.fp8_meta["scaling_bwd"].amax_history[0] for m in self._modules]
+
+    def __enter__(self):
+        self._original = FP8GlobalStateManager.reduce_and_update_quantization_state.__func__
+        original = self._original
+        counter = self
+
+        def counted(cls, forward=True):
+            if not forward:
+                counter.backward += 1
+                for amax in counter._current_bwd_amax():
+                    assert amax.any(), "backward update ran before all modules finished backward"
+            result = original(cls, forward=forward)
+            if not forward:
+                for amax in counter._current_bwd_amax():
+                    assert not amax.any(), "backward update did not roll the amax history"
+            return result
+
+        FP8GlobalStateManager.reduce_and_update_quantization_state = classmethod(counted)
+        return self
+
+    def __exit__(self, *exc):
+        FP8GlobalStateManager.reduce_and_update_quantization_state = classmethod(self._original)
+
+
+def _make_update_test_model(num_layers=3, seed=1234):
+    torch.manual_seed(seed)
+    return torch.nn.ModuleList(
+        [
+            Linear(_UPDATE_TEST_HIDDEN, _UPDATE_TEST_HIDDEN, bias=True).cuda()
+            for _ in range(num_layers)
+        ]
+    )
+
+
+def _run_update_test_layers(layers, x):
+    for layer in layers:
+        x = layer(x)
+    return x
+
+
+def _update_forward_plain(model, x):
+    return _run_update_test_layers(model, x)
+
+
+def _update_forward_reentrant(model, x):
+    return te_checkpoint(_run_update_test_layers, model, x, use_reentrant=True)
+
+
+def _update_forward_non_reentrant(model, x):
+    return te_checkpoint(_run_update_test_layers, model, x, use_reentrant=False)
+
+
+def _update_forward_per_layer_reentrant(model, x):
+    for layer in model:
+        x = te_checkpoint(layer, x, use_reentrant=True)
+    return x
+
+
+def _update_forward_nested(model, x):
+    def inner(value):
+        return te_checkpoint(model[1], value, use_reentrant=True)
+
+    def outer(value):
+        return model[2](inner(model[0](value)))
+
+    return te_checkpoint(outer, x, use_reentrant=True)
+
+
+# Each case builds its modules and returns (step, tracked_modules). ``step``
+# runs one forward + backward; ``tracked_modules`` must all finish backward
+# before the single update of that step.
+
+
+def _update_case_single_graph(forward_fn):
+    def build():
+        model = _make_update_test_model()
+
+        def step(x, recipe, counter):
+            with autocast(enabled=True, recipe=recipe):
+                out = forward_fn(model, x)
+            out.float().sum().backward()
+
+        return step, [model]
+
+    return build
+
+
+def _update_case_branched(checkpoint_first_branch):
+    def build():
+        branch_a = _make_update_test_model(num_layers=2, seed=1)
+        branch_b = _make_update_test_model(num_layers=2, seed=2)
+
+        def step(x, recipe, counter):
+            with autocast(enabled=True, recipe=recipe):
+                if checkpoint_first_branch:
+                    out_a = te_checkpoint(_run_update_test_layers, branch_a, x, use_reentrant=True)
+                else:
+                    out_a = _run_update_test_layers(branch_a, x)
+                out_b = te_checkpoint(_run_update_test_layers, branch_b, x, use_reentrant=True)
+            (out_a + out_b).float().sum().backward()
+
+        return step, [branch_a, branch_b]
+
+    return build
+
+
+def _update_case_unused_checkpoint_branch():
+    used = _make_update_test_model(num_layers=2, seed=1)
+    unused = _make_update_test_model(num_layers=2, seed=2)
+
+    def step(x, recipe, counter):
+        with autocast(enabled=True, recipe=recipe):
+            unused_out = te_checkpoint(_run_update_test_layers, unused, x, use_reentrant=True)
+            out = _run_update_test_layers(used, x)
+        out.float().sum().backward()
+        del unused_out
+
+    return step, [used]
+
+
+def _update_case_scope_independent_graphs():
+    models = [_make_update_test_model(num_layers=2, seed=seed) for seed in (1, 2)]
+
+    def step(x, recipe, counter):
+        with autocast(enabled=True, recipe=recipe):
+            outputs = [_run_update_test_layers(model, x) for model in models]
+        updates_before = counter.backward
+        with quantization_backward_scope():
+            for output in outputs:
+                output.float().sum().backward()
+                assert counter.backward == updates_before
+
+    return step, models
+
+
+def _update_case_scope_delayed_wgrad():
+    model = Linear(
+        _UPDATE_TEST_HIDDEN,
+        _UPDATE_TEST_HIDDEN,
+        bias=True,
+        delay_wgrad_compute=True,
+    ).cuda()
+
+    def step(x, recipe, counter):
+        updates_before = counter.backward
+        with quantization_backward_scope():
+            with autocast(enabled=True, recipe=recipe):
+                out = model(x)
+            out.float().sum().backward()
+            assert counter.backward == updates_before
+            model.backward_dw()
+            assert model.weight.grad is not None
+            assert counter.backward == updates_before
+
+    return step, [model]
+
+
+_UPDATE_TEST_CASES = {
+    "plain": _update_case_single_graph(_update_forward_plain),
+    "reentrant": _update_case_single_graph(_update_forward_reentrant),
+    "non_reentrant": _update_case_single_graph(_update_forward_non_reentrant),
+    "per_layer_reentrant": _update_case_single_graph(_update_forward_per_layer_reentrant),
+    "nested": _update_case_single_graph(_update_forward_nested),
+    "branched": _update_case_branched(checkpoint_first_branch=False),
+    "branched_checkpoint_first": _update_case_branched(checkpoint_first_branch=True),
+    "unused_checkpoint_branch": _update_case_unused_checkpoint_branch,
+    "scope_independent_graphs": _update_case_scope_independent_graphs,
+    "scope_delayed_wgrad": _update_case_scope_delayed_wgrad,
+}
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.parametrize("case", _UPDATE_TEST_CASES.keys())
+def test_delayed_scaling_updates_once_per_backward(case):
+    FP8GlobalStateManager.reset()
+    step, tracked = _UPDATE_TEST_CASES[case]()
+    fp8_recipe = recipe.DelayedScaling()
+
+    with _UpdateCounter(*tracked) as counter:
+        for i in range(_UPDATE_TEST_STEPS):
+            x = torch.randn(
+                _UPDATE_TEST_BATCH,
+                _UPDATE_TEST_HIDDEN,
+                device="cuda",
+                requires_grad=True,
+            )
+            step(x, fp8_recipe, counter)
+            assert counter.backward == i + 1
+            assert x.grad is not None and torch.isfinite(x.grad).all()
+            qstate = FP8GlobalStateManager.quantization_state
+            assert not qstate.pending_backward_quantization_update
+            assert qstate.backward_quantization_update_callback_task_id is None
+
+
 def _test_e2e_checkpointing_get_model(config, dtype):
     sigma = 0.023
     init_method = init_method_normal(sigma)
diff --git a/transformer_engine/pytorch/__init__.py b/transformer_engine/pytorch/__init__.py
index 2b1803bfb2..286eef2561 100644
--- a/transformer_engine/pytorch/__init__.py
+++ b/transformer_engine/pytorch/__init__.py
@@ -45,6 +45,7 @@
 from transformer_engine.pytorch.quantization import fp8_autocast
 from transformer_engine.pytorch.quantization import fp8_model_init
 from transformer_engine.pytorch.quantization import autocast
+from transformer_engine.pytorch.quantization import quantization_backward_scope
 from transformer_engine.pytorch.quantization import quantized_model_init
 from transformer_engine.pytorch.quantization import is_fp8_available
 from transformer_engine.pytorch.quantization import is_mxfp8_available
diff --git a/transformer_engine/pytorch/distributed.py b/transformer_engine/pytorch/distributed.py
index 8605a4746b..a2d5a621b6 100644
--- a/transformer_engine/pytorch/distributed.py
+++ b/transformer_engine/pytorch/distributed.py
@@ -39,7 +39,7 @@
 )
 
 from .constants import dist_group_type
-from .quantization import FP8GlobalStateManager, autocast
+from .quantization import FP8GlobalStateManager, autocast, quantization_backward_scope
 from .tensor.float8_tensor import Float8Quantizer, Float8Tensor, Float8CurrentScalingQuantizer
 from .tensor.mxfp8_tensor import MXFP8Quantizer
 from .tensor.nvfp4_tensor import NVFP4Quantizer
@@ -247,8 +247,6 @@ class activation_recompute_forward(AbstractContextManager, ContextDecorator):
     activations, followed by calculation of gradients using these values.
     """
 
-    _is_first_fp8_module: List = []
-
     def __init__(self, activation_recompute: bool = False, recompute_phase: bool = False):
         super().__init__()
         self.activation_recompute = activation_recompute
@@ -264,12 +262,6 @@ def __enter__(self):
         _IN_ACTIVATION_RECOMPUTE_REGION = self.activation_recompute
         _ACTIVATION_RECOMPUTE_PHASE = self.recompute_phase
 
-        qstate = FP8GlobalStateManager.quantization_state
-        if self.activation_recompute and not self.recompute_phase:
-            activation_recompute_forward._is_first_fp8_module.append(qstate.is_first_fp8_module)
-        if self.activation_recompute and self.recompute_phase:
-            qstate.is_first_fp8_module = activation_recompute_forward._is_first_fp8_module.pop(0)
-
     def __exit__(self, *exc_details):
         global _IN_ACTIVATION_RECOMPUTE_REGION, _ACTIVATION_RECOMPUTE_PHASE
         _IN_ACTIVATION_RECOMPUTE_REGION = False
@@ -407,6 +399,20 @@ def backward(
         ctx, *args: Tuple[Union[torch.Tensor, None], ...]
     ) -> Tuple[Union[torch.Tensor, None], ...]:
         """Call backward function with activation recomputation."""
+        recipe = ctx.fp8_recipe
+        update_scope = (
+            quantization_backward_scope()
+            if ctx.fp8 and (recipe.delayed() or recipe.custom())
+            else nullcontext()
+        )
+        with update_scope:
+            return _CheckpointFunction._backward(ctx, *args)
+
+    @staticmethod
+    def _backward(
+        ctx, *args: Tuple[Union[torch.Tensor, None], ...]
+    ) -> Tuple[Union[torch.Tensor, None], ...]:
+        """Recompute the forward and run its nested backward."""
         if not torch.autograd._is_checkpoint_valid():
             raise RuntimeError(
                 "Checkpointing is not compatible with .grad(), please use .backward() if possible"
diff --git a/transformer_engine/pytorch/module/grouped_linear.py b/transformer_engine/pytorch/module/grouped_linear.py
index 612a430966..f62d500d0f 100644
--- a/transformer_engine/pytorch/module/grouped_linear.py
+++ b/transformer_engine/pytorch/module/grouped_linear.py
@@ -40,7 +40,6 @@
     clear_tensor_data,
     get_device_compute_capability,
     init_method_constant,
-    requires_grad,
     resolve_grouped_linear_single_param_flags,
     get_nvtx_range_context,
 )
@@ -599,12 +598,9 @@ def _forward_grouped_tensor(
             ctx.use_bias = use_bias
             ctx.inp_shape = inp.shape
             ctx.requires_dgrad = inp.requires_grad
-            ctx.reduce_and_update_bwd_fp8_tensors = False
-            if ctx.fp8 and requires_grad(inp, weights[0], biases[0]):
-                ctx.reduce_and_update_bwd_fp8_tensors = (
-                    ctx.reduce_and_update_bwd_fp8_tensors
-                    or FP8GlobalStateManager.is_first_fp8_module()
-                )
+            ctx.request_backward_quantization_update = (
+                ctx.fp8 and FP8GlobalStateManager.backward_quantization_update_needed()
+            )
             ctx.wgrad_store = wgrad_store
             ctx.debug = False
             ctx.save_original_input = save_original_input
@@ -975,12 +971,6 @@ def forward(
             ctx.sequence_parallel = sequence_parallel
             ctx.inp_shape = inp.shape
             ctx.requires_dgrad = inp.requires_grad
-            ctx.reduce_and_update_bwd_fp8_tensors = False
-            if ctx.fp8 and requires_grad(inp, weights[0], biases[0]):
-                ctx.reduce_and_update_bwd_fp8_tensors = (
-                    ctx.reduce_and_update_bwd_fp8_tensors
-                    or FP8GlobalStateManager.is_first_fp8_module()
-                )
             ctx.wgrad_store = wgrad_store
             ctx.debug = debug
             ctx.save_original_input = save_original_input
@@ -998,7 +988,9 @@ def forward(
                 ctx.grad_input_quantizers = [None] * num_gemms
                 ctx.grad_weight_quantizers = [None] * num_gemms
                 ctx.grad_output_quantizers = [None] * num_gemms
-                ctx.reduce_and_update_bwd_fp8_tensors = False
+            ctx.request_backward_quantization_update = (
+                ctx.fp8 and FP8GlobalStateManager.backward_quantization_update_needed()
+            )
 
         # [*, in_features] -> [*, out_features] except first dimension changes for SP
         return out.view(-1, *inp.shape[1:-1], out.shape[-1]), new_workspaces
@@ -1250,8 +1242,8 @@ def handle_custom_ddp_from_mcore(weight, main_grad, wgrad):
         else:
             wgrad_list = [None] * num_weight_args
 
-        if ctx.reduce_and_update_bwd_fp8_tensors:
-            FP8GlobalStateManager.reduce_and_update_fp8_tensors(forward=False)
+        if ctx.request_backward_quantization_update:
+            FP8GlobalStateManager.request_backward_quantization_update()
         return (
             dgrad.view(ctx.inp_shape) if ctx.requires_dgrad else None,
             None,  # m_splits
@@ -1517,8 +1509,8 @@ def handle_custom_ddp_from_mcore(weight, main_grad, wgrad):
             ):
                 grad_biases = [None] * ctx.num_gemms
 
-        if ctx.reduce_and_update_bwd_fp8_tensors:
-            FP8GlobalStateManager.reduce_and_update_fp8_tensors(forward=False)
+        if ctx.request_backward_quantization_update:
+            FP8GlobalStateManager.request_backward_quantization_update()
         return (
             dgrad.view(ctx.inp_shape) if ctx.requires_dgrad else None,
             None,  # m_splits
diff --git a/transformer_engine/pytorch/module/layernorm_linear.py b/transformer_engine/pytorch/module/layernorm_linear.py
index 561e813348..13dc428161 100644
--- a/transformer_engine/pytorch/module/layernorm_linear.py
+++ b/transformer_engine/pytorch/module/layernorm_linear.py
@@ -41,7 +41,6 @@
     init_method_constant,
     nvtx_range_pop,
     nvtx_range_push,
-    requires_grad,
     needs_quantized_gemm,
     get_nvtx_range_context,
 )
@@ -52,7 +51,6 @@
     symmetric_all_reduce,
     reduce_scatter_along_first_dim,
     gather_along_first_dim,
-    in_fp8_activation_recompute_phase,
     _fsdp_scatter_tensors,
     _fsdp_gather_tensors,
 )
@@ -64,7 +62,6 @@
 )
 from ..constants import FP8BwdTensorIdx, FP8FwdTensorIdx, GemmParallelModes, dist_group_type
 from ..jit import no_torch_dynamo
-from ..graph import is_graph_capturing
 from ._common import (
     apply_normalization,
     noop_cat,
@@ -588,13 +585,6 @@ def forward(
             ctx.ub_name = ub_name
             ctx.requires_dgrad = inp_requires_grad
             ctx.normalization = normalization
-            ctx.reduce_and_update_bwd_fp8_tensors = False
-            if ctx.fp8 and requires_grad(inp, ln_weight, ln_bias, weight, bias):
-                qstate = FP8GlobalStateManager.quantization_state
-                _first_fp8_module = qstate.is_first_fp8_module
-                ctx.reduce_and_update_bwd_fp8_tensors = FP8GlobalStateManager.is_first_fp8_module()
-                if in_fp8_activation_recompute_phase():
-                    qstate.is_first_fp8_module = _first_fp8_module
             ctx.wgrad_store = wgrad_store
             ctx.debug = debug
 
@@ -609,7 +599,9 @@ def forward(
                 ctx.grad_input_quantizer = None
                 ctx.grad_weight_quantizer = None
                 ctx.grad_output_quantizer = None
-                ctx.reduce_and_update_bwd_fp8_tensors = False
+            ctx.request_backward_quantization_update = (
+                ctx.fp8 and FP8GlobalStateManager.backward_quantization_update_needed()
+            )
 
         # ------------------------------------------------------
         # Cached state for backward pass is ready...
@@ -1184,10 +1176,8 @@ def wgrad_gemm(
         else:
             wgrad = None
 
-        if ctx.reduce_and_update_bwd_fp8_tensors and not is_graph_capturing():
-            nvtx_range_push(f"{nvtx_label}.reduce_and_update_fp8_tensors")
-            FP8GlobalStateManager.reduce_and_update_fp8_tensors(forward=False)
-            nvtx_range_pop(f"{nvtx_label}.reduce_and_update_fp8_tensors")
+        if ctx.request_backward_quantization_update:
+            FP8GlobalStateManager.request_backward_quantization_update()
 
         # Scatter fp8 weight buffers
         # if ctx.fp8 and not isinstance(weight, QuantizedTensorStorage):
diff --git a/transformer_engine/pytorch/module/layernorm_mlp.py b/transformer_engine/pytorch/module/layernorm_mlp.py
index 3ee0cda50c..da9c1d3cc5 100644
--- a/transformer_engine/pytorch/module/layernorm_mlp.py
+++ b/transformer_engine/pytorch/module/layernorm_mlp.py
@@ -46,7 +46,6 @@
     cast_if_needed,
     assert_dim_for_fp8_exec,
     clear_tensor_data,
-    requires_grad,
     needs_quantized_gemm,
     get_nvtx_range_context,
 )
@@ -58,14 +57,12 @@
     reduce_scatter_along_first_dim,
     gather_along_first_dim,
     use_reentrant_activation_recompute,
-    in_fp8_activation_recompute_phase,
     _fsdp_scatter_tensors,
     _get_cuda_rng_state,
     _set_cuda_rng_state,
 )
 from ..constants import FP8BwdTensorIdx, FP8FwdTensorIdx, dist_group_type
 from ..jit import no_torch_dynamo
-from ..graph import is_graph_capturing
 from ..tensor.float8_tensor import Float8Tensor
 from ..tensor.mxfp8_tensor import MXFP8Quantizer
 from ..tensor.nvfp4_tensor import NVFP4Quantizer
@@ -908,17 +905,10 @@ def _forward(
                 inp.requires_grad or ln_weight.requires_grad or ln_bias.requires_grad
             )
             ctx.normalization = normalization
-            ctx.reduce_and_update_bwd_fp8_tensors = False
-            if ctx.fp8 and requires_grad(
-                inp, ln_weight, ln_bias, fc1_weight, fc2_weight, fc1_bias, fc2_bias
-            ):
-                qstate = FP8GlobalStateManager.quantization_state
-                _first_fp8_module = qstate.is_first_fp8_module
-                ctx.reduce_and_update_bwd_fp8_tensors = FP8GlobalStateManager.is_first_fp8_module()
-                if in_fp8_activation_recompute_phase() or is_recomputation:
-                    qstate.is_first_fp8_module = _first_fp8_module
-
             ctx.wgrad_store = wgrad_store
+            ctx.request_backward_quantization_update = (
+                ctx.fp8 and FP8GlobalStateManager.backward_quantization_update_needed()
+            )
             if is_recomputation:  # return the recomputed tensors
                 return (
                     ctx,
@@ -1806,8 +1796,8 @@ def fc1_wgrad_gemm(
         else:
             fc2_wgrad = None
 
-        if ctx.reduce_and_update_bwd_fp8_tensors and not is_graph_capturing():
-            FP8GlobalStateManager.reduce_and_update_fp8_tensors(forward=False)
+        if ctx.request_backward_quantization_update:
+            FP8GlobalStateManager.request_backward_quantization_update()
 
         # FIX THIS
         # Scatter Fp8 tranposed-weight buffers
diff --git a/transformer_engine/pytorch/module/linear.py b/transformer_engine/pytorch/module/linear.py
index 56622db5e6..16d8bb2d3d 100644
--- a/transformer_engine/pytorch/module/linear.py
+++ b/transformer_engine/pytorch/module/linear.py
@@ -72,7 +72,6 @@
 )
 from ..constants import FP8BwdTensorIdx, FP8FwdTensorIdx, GemmParallelModes, dist_group_type
 from ..jit import no_torch_dynamo
-from ..graph import is_graph_capturing
 from ..quantized_tensor import (
     QuantizedTensor,
     QuantizedTensorStorage,
@@ -198,6 +197,7 @@ class LinearBwdArgs:
     # --- Numerical / dtype config ---
     activation_dtype: Optional[torch.dtype] = None
     fp8: bool = False
+    request_backward_quantization_update: bool = False
     dgrad_use_split_accumulator: bool = _2X_ACC_DGRAD
     wgrad_use_split_accumulator: bool = _2X_ACC_WGRAD
     backward_override: Optional[str] = None
@@ -233,9 +233,6 @@ class LinearBwdArgs:
     origin_weight_overwrites_main_grad: bool = False
     main_grad_func: Optional[Callable[[], torch.Tensor]] = None
 
-    # --- FP8 reduce-and-update bookkeeping ---
-    reduce_and_update_bwd_fp8_tensors: bool = False
-
     # --- Misc ---
     cpu_offloading: bool = False
     owns_input: bool = False
@@ -255,16 +252,6 @@ def setup_saved_tensors(self, ctx: torch.autograd.function.FunctionCtx) -> None:
         )  # pylint: disable=unbalanced-tuple-unpacking
 
 
-def _check_fp8_reduce_and_update():
-    """Check if this is the first FP8 module (for backward reduce-and-update)."""
-    qstate = FP8GlobalStateManager.quantization_state
-    _first_fp8_module = qstate.is_first_fp8_module
-    result = FP8GlobalStateManager.is_first_fp8_module()
-    if in_fp8_activation_recompute_phase():
-        qstate.is_first_fp8_module = _first_fp8_module
-    return result
-
-
 def _linear_forward_impl(
     args: LinearFwdArgs,
 ) -> Tuple[torch.Tensor, Optional[torch.Tensor], Optional[Tuple], None, Optional[Dict]]:
@@ -769,6 +756,9 @@ def _linear_setup_ctx(
         bwd_args.grad_input_quantizer = None
         bwd_args.grad_weight_quantizer = None
         bwd_args.grad_output_quantizer = None
+    bwd_args.request_backward_quantization_update = (
+        bwd_args.fp8 and FP8GlobalStateManager.backward_quantization_update_needed()
+    )
 
     saved_inputmat, wt_save, saved_weight, saved_bias = tensors_to_save_from_forward
     inputmat_alias, wt_save_alias, saved_weight_alias, bias_alias = ctx_attrs[
@@ -1434,14 +1424,6 @@ def forward(
             ctx.save_for_backward(*tensors_to_save)
             ctx.tensor_objects = tensor_objects
             ctx.backward_objects = bwd_args
-            if fwd_args.fp8 and (
-                fwd_args.input_requires_grad
-                or fwd_args.weight_requires_grad
-                or fwd_args.bias_requires_grad
-            ):
-                bwd_args.reduce_and_update_bwd_fp8_tensors = _check_fp8_reduce_and_update()
-            if fwd_args.backward_override is not None:
-                bwd_args.reduce_and_update_bwd_fp8_tensors = False
 
         return out, new_weight_workspace
 
@@ -1459,15 +1441,13 @@ def backward(
         if bwd_args.ub_name is not None:
             nvtx_label = f"{nvtx_label}.{bwd_args.ub_name}"
         result = _linear_backward(bwd_args) + (None,)  # fwd_args grad slot
-        reduce_and_update_bwd_fp8_tensors = bwd_args.reduce_and_update_bwd_fp8_tensors
+        request_update = bwd_args.request_backward_quantization_update
         # Drop all references held by bwd_args (saved tensors, quantizers, weakrefs,
         # main_grad closure) so they don't outlive backward via ctx under retain_graph.
         ctx.backward_objects = None
         del bwd_args
-        if reduce_and_update_bwd_fp8_tensors and not is_graph_capturing():
-            nvtx_range_push(f"{nvtx_label}.reduce_and_update_fp8_tensors")
-            FP8GlobalStateManager.reduce_and_update_fp8_tensors(forward=False)
-            nvtx_range_pop(f"{nvtx_label}.reduce_and_update_fp8_tensors")
+        if request_update:
+            FP8GlobalStateManager.request_backward_quantization_update()
         return result
 
 
diff --git a/transformer_engine/pytorch/ops/fuser.py b/transformer_engine/pytorch/ops/fuser.py
index fd66529ba8..7dac79810c 100644
--- a/transformer_engine/pytorch/ops/fuser.py
+++ b/transformer_engine/pytorch/ops/fuser.py
@@ -5,7 +5,7 @@
 """Manager class for a pipeline of fusible operations."""
 
 from __future__ import annotations
-from collections.abc import Callable, Iterable, Sequence
+from collections.abc import Iterable, Sequence
 import itertools
 from typing import Any, Optional, TypeAlias
 
@@ -26,24 +26,6 @@ def _split_tuple(t: tuple, idx: int) -> tuple[tuple, tuple]:
     return t[:idx], t[idx:]
 
 
-# Lazily imported function used in _is_graph_capturing
-_is_graph_capturing_function: Optional[Callable[[], bool]] = None
-
-
-def _is_graph_capturing() -> bool:
-    """Whether function is called within ``make_graphed_callables``
-
-    Avoid circular import with lazy import.
-
-    """
-    global _is_graph_capturing_function
-    if _is_graph_capturing_function is None:
-        from ..graph import is_graph_capturing
-
-        _is_graph_capturing_function = is_graph_capturing
-    return _is_graph_capturing_function()
-
-
 # Type alias for a function that may perform operation fusion
 OperationFusionFunction: TypeAlias = (
     "Callable[tuple[list[FusibleOperation], ...], list[FusibleOperation]]"
@@ -227,11 +209,6 @@ def forward(
             func_ctx.save_for_backward(*tensors_to_save)
             func_ctx.tensor_objects = tensor_objects
 
-            # Whether to perform recipe update in backward pass
-            is_first_module = False
-            if fuser.first_op_requiring_backward < fuser._num_basic_ops:
-                is_first_module = FP8GlobalStateManager.is_first_fp8_module()
-
             # Other context
             func_ctx.backward_ops = fuser._backward_ops
             func_ctx.basic_ops = fuser._basic_ops
@@ -243,7 +220,9 @@ def forward(
             func_ctx.basic_op_extra_output_channels = fuser._basic_op_extra_output_channels
             func_ctx.basic_op_extra_output_consumers = fuser._basic_op_extra_output_consumers
             func_ctx.basic_op_extra_input_sources = fuser._basic_op_extra_input_sources
-            func_ctx.is_first_module = is_first_module
+            func_ctx.request_backward_quantization_update = (
+                FP8GlobalStateManager.backward_quantization_update_needed()
+            )
 
         # Mark output tensors as not deletable in backward
         for tensor in itertools.chain(
@@ -383,9 +362,8 @@ def backward(
             for op_idx, input_idx in func_ctx.external_extra_input_slots
         ]
 
-        # Update FP8 scaling factors
-        if func_ctx.is_first_module and not _is_graph_capturing():
-            FP8GlobalStateManager.reduce_and_update_fp8_tensors(forward=False)
+        if func_ctx.request_backward_quantization_update:
+            FP8GlobalStateManager.request_backward_quantization_update()
 
         return (
             dx,  # input_
diff --git a/transformer_engine/pytorch/quantization.py b/transformer_engine/pytorch/quantization.py
index 8c7ea22263..02fd2cae39 100644
--- a/transformer_engine/pytorch/quantization.py
+++ b/transformer_engine/pytorch/quantization.py
@@ -28,12 +28,13 @@
 )
 from .constants import dist_group_type, DType
 
-from .utils import get_device_compute_capability
+from .utils import get_device_compute_capability, nvtx_range_push, nvtx_range_pop
 from .jit import jit_fuser
 
 
 __all__ = [
     "autocast",
+    "quantization_backward_scope",
     "quantized_model_init",
     "is_fp8_available",
     "is_mxfp8_available",
@@ -400,6 +401,9 @@ class FP8GlobalState:
     fp8_parameters: bool = False
     high_precision_init_val: bool = False
     is_first_fp8_module: bool = False
+    pending_backward_quantization_update: bool = False
+    backward_quantization_update_callback_task_id: Optional[int] = None
+    quantization_backward_scope_depth: int = 0
     fp8_graph_capturing: bool = False
     autocast_depth: int = 0
     global_amax_buffer: Dict[str, list] = field(default_factory=dict)
@@ -653,7 +657,7 @@ def reduce_tensor_across_group_op_max(tensor: torch.Tensor, group: dist_group_ty
             )
 
     @classmethod
-    def reduce_and_update_fp8_tensors(
+    def reduce_and_update_quantization_state(
         cls,
         forward: bool = True,
     ) -> None:
@@ -680,6 +684,83 @@ def reduce_and_update_fp8_tensors(
                 qstate.global_scale_buffer[buffer_key],
             )
 
+    # Compatibility alias used by Megatron-Core.
+    reduce_and_update_fp8_tensors = reduce_and_update_quantization_state
+
+    @classmethod
+    def backward_quantization_update_needed(cls) -> bool:
+        """Whether modules in the active autocast must request an update after backward."""
+        if not cls.is_fp8_enabled():
+            return False
+        recipe = cls.get_fp8_recipe()
+        return recipe.delayed() or recipe.custom()
+
+    @classmethod
+    def request_backward_quantization_update(cls) -> None:
+        """Request an update after the enclosing logical backward.
+
+        No-op inside CUDA graph capture, where the graphed wrapper performs the update.
+        """
+        from .graph import is_graph_capturing  # pylint: disable=import-outside-toplevel
+
+        if is_graph_capturing():
+            return
+        qstate = cls.quantization_state
+        qstate.pending_backward_quantization_update = True
+        if qstate.quantization_backward_scope_depth == 0:
+            cls._queue_backward_quantization_update_callback()
+
+    @classmethod
+    def _queue_backward_quantization_update_callback(cls, task_id: Optional[int] = None) -> None:
+        """Queue an update after an autograd task."""
+        qstate = cls.quantization_state
+        if task_id is None:
+            task_id = torch._C._current_graph_task_id()
+        if task_id == -1:
+            raise RuntimeError("Backward quantization update must be requested during backward")
+        if qstate.backward_quantization_update_callback_task_id == task_id:
+            return
+
+        qstate.backward_quantization_update_callback_task_id = task_id
+
+        def callback() -> None:
+            cls._run_backward_quantization_update_callback(task_id)
+
+        try:
+            torch.autograd.Variable._execution_engine.queue_callback(callback)
+        except RuntimeError:
+            if qstate.backward_quantization_update_callback_task_id == task_id:
+                qstate.backward_quantization_update_callback_task_id = None
+            raise
+
+    @classmethod
+    def _run_backward_quantization_update_callback(cls, task_id: int) -> None:
+        """Run the update callback for an autograd task."""
+        qstate = cls.quantization_state
+        if qstate.backward_quantization_update_callback_task_id != task_id:
+            return
+        qstate.backward_quantization_update_callback_task_id = None
+        if qstate.quantization_backward_scope_depth == 0:
+            cls._run_pending_backward_quantization_update()
+
+    @classmethod
+    def _run_pending_backward_quantization_update(cls) -> None:
+        """Run the pending backward update, if any."""
+        qstate = cls.quantization_state
+        if not qstate.pending_backward_quantization_update:
+            return
+        qstate.pending_backward_quantization_update = False
+        nvtx_range_push("transformer_engine.reduce_and_update_quantization_state.backward")
+        update_succeeded = False
+        try:
+            with torch.no_grad():
+                cls.reduce_and_update_quantization_state(forward=False)
+            update_succeeded = True
+        finally:
+            if not update_succeeded:
+                qstate.pending_backward_quantization_update = True
+            nvtx_range_pop("transformer_engine.reduce_and_update_quantization_state.backward")
+
     @staticmethod
     def get_unique_autocast_key(
         recipe: Optional[Recipe] = None,
@@ -749,7 +830,7 @@ def autocast_exit(cls, enabled: bool, _graph: bool) -> None:
         if enabled and qstate.autocast_depth == 0 and not _graph and torch.is_grad_enabled():
             # delayed scaling only function, for other recipes (current scaling with any granularity),
             # this is noop for other recipes because cls.global_amax_buffer is empty list
-            cls.reduce_and_update_fp8_tensors(forward=True)
+            cls.reduce_and_update_quantization_state(forward=True)
 
     @classmethod
     def copy_forward_fp8_meta_tensors_for_recompute(cls, fp8_meta: Dict[str, Any]) -> None:
@@ -813,6 +894,50 @@ def restore_fp8_meta_tensors(fp8_meta: Dict[str, Any]) -> None:
         fp8_meta["scaling_fwd"].scale.copy_(fp8_meta["updated_scale_fwd"])
 
 
+@contextmanager
+def quantization_backward_scope() -> None:
+    """Run the delayed-scaling update once, at the end of a logical backward.
+
+    With delayed scaling, each backward pass ends by reducing the gradient amaxes
+    across the amax reduction group and recomputing the scales. Ordinary
+    ``loss.backward()`` calls do this automatically and do not need this scope.
+
+    Use it when one training step consists of several ``backward()`` calls, so that
+    the update runs once when the outermost scope exits instead of after every call:
+
+    .. code-block:: python
+
+        with te.quantization_backward_scope():
+            for loss in microbatch_losses:  # e.g. a 1F1B pipeline schedule
+                loss.backward()
+
+    Nested scopes are no-ops; only the outermost one triggers the update. A scope
+    entered inside a running backward (e.g. in a hook) defers the update to the
+    end of that backward.
+
+    The update also runs if no quantized module ran backward inside the scope, so
+    a rank that skipped all of them (e.g. an expert that received no tokens) still
+    joins the amax reduction and does not stall the other ranks. Every rank must
+    therefore enter and exit the scope the same number of times per step.
+
+    For recipes without delayed-scaling state this scope has no effect.
+    """
+    qstate = FP8GlobalStateManager.quantization_state
+    outermost = qstate.quantization_backward_scope_depth == 0
+    if outermost:
+        qstate.pending_backward_quantization_update = True
+    task_id = torch._C._current_graph_task_id() if outermost else -1
+    if task_id != -1:
+        FP8GlobalStateManager._queue_backward_quantization_update_callback(task_id)
+    qstate.quantization_backward_scope_depth += 1
+    try:
+        yield
+    finally:
+        qstate.quantization_backward_scope_depth -= 1
+        if outermost and task_id == -1:
+            FP8GlobalStateManager._run_pending_backward_quantization_update()
+
+
 @contextmanager
 def fp8_model_init(
     enabled: bool = True,