diff --git a/docs/api/pytorch.rst b/docs/api/pytorch.rst
index 5fac0a89a6..3115fb027f 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.backward_quantization_update_scope
+
 .. autoapifunction:: transformer_engine.pytorch.quantized_model_init
 
 .. autoapifunction:: transformer_engine.pytorch.checkpoint
diff --git a/tests/pytorch/test_backward_override.py b/tests/pytorch/test_backward_override.py
index c0acf2e6b3..44d88bc9f7 100644
--- a/tests/pytorch/test_backward_override.py
+++ b/tests/pytorch/test_backward_override.py
@@ -419,7 +419,7 @@ def _snapshot_backward_ctx_state(
         "backward_override",
         "fp8",
         "grad_output_quantizer",
-        "reduce_and_update_bwd_fp8_tensors",
+        "should_request_backward_quantization_update",
     )
     missing_attrs = [attr for attr in required_attrs if not hasattr(state_holder, attr)]
     if missing_attrs:
@@ -430,7 +430,7 @@ 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")),
+        bool(getattr(state_holder, "should_request_backward_quantization_update")),
     )
 
 
@@ -816,7 +816,7 @@ def _run_grouped_linear_single_step_with_ctx_state(
         required_attrs = (
             "backward_override",
             "fp8",
-            "reduce_and_update_bwd_fp8_tensors",
+            "should_request_backward_quantization_update",
         )
         missing_attrs = [attr for attr in required_attrs if not hasattr(y.grad_fn, attr)]
         if missing_attrs:
@@ -827,7 +827,7 @@ 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")),
+            bool(getattr(y.grad_fn, "should_request_backward_quantization_update")),
         )
     y.backward(dy)
     assert x_run.grad is not None
diff --git a/tests/pytorch/test_fusible_ops.py b/tests/pytorch/test_fusible_ops.py
index 3a0ad7dd3f..8480731f8a 100644
--- a/tests/pytorch/test_fusible_ops.py
+++ b/tests/pytorch/test_fusible_ops.py
@@ -25,6 +25,7 @@
 )
 from transformer_engine.pytorch.ops.fuser import OperationFuser
 from transformer_engine.pytorch._extra_state import UNSAFE_PICKLE_EXTRA_STATE_ENV
+from transformer_engine.pytorch.quantization import FP8GlobalStateManager
 
 from transformer_engine.pytorch.ops.fused import (
     BackwardActivationBias,
@@ -1030,6 +1031,7 @@ def test_fp8_scale_update(
             with te.autocast(recipe=recipe):
                 y = model(x)
             y.backward(dy)
+            FP8GlobalStateManager.flush_backward_quantization_update()
             with torch.no_grad():
                 model.weight.fill_(w_vals[step + 1])
 
diff --git a/tests/pytorch/test_numerics.py b/tests/pytorch/test_numerics.py
index e6a83d92bc..6324ab5f11 100644
--- a/tests/pytorch/test_numerics.py
+++ b/tests/pytorch/test_numerics.py
@@ -43,8 +43,10 @@
 )
 from transformer_engine.pytorch import checkpoint as te_checkpoint
 from transformer_engine.pytorch.distributed import (
-    is_fp8_activation_recompute_enabled,
+    activation_recompute_forward,
+    in_fp8_activation_recompute_forward_phase,
     in_fp8_activation_recompute_phase,
+    is_fp8_activation_recompute_enabled,
 )
 from transformer_engine.pytorch.cpp_extensions import general_gemm
 from transformer_engine.common import recipe
@@ -906,6 +908,23 @@ def _checkpointed_linear_backward(body, use_reentrant, *layers):
         assert torch.isfinite(layer.weight.grad).all()
 
 
+def test_nested_activation_recompute_phases():
+    """Nested checkpoint forwards preserve an active outer recompute phase."""
+    FP8GlobalStateManager.reset()
+
+    with activation_recompute_forward(True, True):
+        assert in_fp8_activation_recompute_phase()
+        assert not in_fp8_activation_recompute_forward_phase()
+        with activation_recompute_forward(True, False):
+            assert in_fp8_activation_recompute_phase()
+            assert in_fp8_activation_recompute_forward_phase()
+        assert in_fp8_activation_recompute_phase()
+        assert not in_fp8_activation_recompute_forward_phase()
+
+    assert not in_fp8_activation_recompute_phase()
+    assert not in_fp8_activation_recompute_forward_phase()
+
+
 @pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
 @pytest.mark.parametrize("use_reentrant", all_boolean)
 def test_checkpoint_inner_autocast_is_an_fp8_recompute_region(use_reentrant):
diff --git a/tests/pytorch/test_recipe.py b/tests/pytorch/test_recipe.py
index ccef104a33..990d50b19e 100644
--- a/tests/pytorch/test_recipe.py
+++ b/tests/pytorch/test_recipe.py
@@ -32,6 +32,7 @@
     _amax_and_scale_update,
 )
 import transformer_engine.pytorch.ops as te_ops
+from transformer_engine.pytorch.distributed import checkpoint as te_checkpoint
 from transformer_engine.common.recipe import (
     CustomRecipe,
     DelayedScaling,
@@ -94,6 +95,7 @@ def test_fp8_scale_update_with_linear_module(
                 is_first_microbatch=True,
             )
         y.backward(torch.zeros_like(y))
+        FP8GlobalStateManager.flush_backward_quantization_update()
 
         # Get amax history and scaling factors
         fp8_meta = module.fp8_meta
@@ -147,6 +149,7 @@ def test_fp8_scale_update_with_linear_module(
             x = torch.randn([16, 16], device="cuda")
             y = module(x, is_first_microbatch=is_first_microbatch)
         y.backward(torch.randn_like(y))
+        FP8GlobalStateManager.flush_backward_quantization_update()
 
         # Check that amax history matches expected values
         torch.testing.assert_close(
@@ -245,6 +248,7 @@ def test_fp8_scale_update_with_linear_fuser_op(
             with te.autocast(recipe=recipe):
                 y = op(x)
             y.backward(dy)
+            FP8GlobalStateManager.flush_backward_quantization_update()
 
             def check_metas(
                 test_scale: float,
@@ -780,3 +784,325 @@ def test_stateful_unknown_or_malformed_pickled_extra_state_requires_opt_in(paylo
 
     monkeypatch.setenv(UNSAFE_PICKLE_EXTRA_STATE_ENV, "1")
     assert should_load_extra_state_pickle(payload, "test")
+
+
+###############################################################################
+# Backward quantization state update scheduling
+###############################################################################
+
+
+HIDDEN = 128
+BATCH = 32
+STEPS = 3
+
+
+class _UpdateCounter:
+    """Counts reduce_and_update_quantization_state calls by direction."""
+
+    def __init__(self):
+        self.forward = 0
+        self.backward = 0
+        self._original = None
+
+    def __enter__(self):
+        self._original = FP8GlobalStateManager.reduce_and_update_quantization_state.__func__
+        original = self._original
+        counter = self
+
+        def counted(cls, forward=True):
+            if forward:
+                counter.forward += 1
+            else:
+                counter.backward += 1
+            return original(cls, forward=forward)
+
+        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_model(num_layers=3, seed=1234):
+    torch.manual_seed(seed)
+    return torch.nn.ModuleList(
+        [te.Linear(HIDDEN, HIDDEN, bias=True).cuda() for _ in range(num_layers)]
+    )
+
+
+def _run_layers(layers, x):
+    for layer in layers:
+        x = layer(x)
+    return x
+
+
+def _train_step(model, x, forward_fn, recipe):
+    with te.autocast(enabled=True, recipe=recipe):
+        out = forward_fn(model, x)
+    loss = out.float().sum()
+    loss.backward()
+    return loss
+
+
+def _forward_plain(model, x):
+    return _run_layers(model, x)
+
+
+def _forward_reentrant(model, x):
+    return te_checkpoint(_run_layers, model, x, use_reentrant=True)
+
+
+def _forward_non_reentrant(model, x):
+    return te_checkpoint(_run_layers, model, x, use_reentrant=False)
+
+
+def _forward_per_layer_reentrant(model, x):
+    for layer in model:
+        x = te_checkpoint(layer, x, use_reentrant=True)
+    return x
+
+
+def _forward_per_layer_non_reentrant(model, x):
+    for layer in model:
+        x = te_checkpoint(layer, x, use_reentrant=False)
+    return x
+
+
+def _forward_nested(model, x):
+    def inner(x):
+        return te_checkpoint(model[1], x, use_reentrant=True)
+
+    def outer(x):
+        x = model[0](x)
+        x = inner(x)
+        return model[2](x)
+
+    return te_checkpoint(outer, x, use_reentrant=True)
+
+
+FORWARD_FNS = {
+    "plain": _forward_plain,
+    "reentrant": _forward_reentrant,
+    "non_reentrant": _forward_non_reentrant,
+    "per_layer_reentrant": _forward_per_layer_reentrant,
+    "per_layer_non_reentrant": _forward_per_layer_non_reentrant,
+    "nested": _forward_nested,
+}
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.parametrize("mode", FORWARD_FNS.keys())
+def test_single_update_per_step(mode):
+    """Exactly one forward and one backward state update per training step."""
+    forward_fn = FORWARD_FNS[mode]
+    model = _make_model()
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        for step in range(STEPS):
+            x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+            _train_step(model, x, forward_fn, recipe)
+            # The backward update of this step is flushed at the next
+            # top-level autocast entry; flush explicitly to count it here.
+            FP8GlobalStateManager.flush_backward_quantization_update()
+            assert counter.forward == step + 1, f"{mode}: duplicate/missing forward update"
+            assert counter.backward == step + 1, f"{mode}: duplicate/missing backward update"
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.skipif(
+    hasattr(torch.autograd.graph, "queue_callback"),
+    reason="public autograd callback API is available",
+)
+def test_flush_at_next_autocast_entry_fallback():
+    """Without an explicit flush, the update runs when the next autocast begins."""
+    model = _make_model()
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+        _train_step(model, x, _forward_plain, recipe)
+        assert counter.backward == 0
+        x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+        _train_step(model, x, _forward_plain, recipe)
+        assert counter.backward == 1
+    FP8GlobalStateManager.flush_backward_quantization_update()
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.skipif(
+    not hasattr(torch.autograd.graph, "queue_callback"),
+    reason="public autograd callback API is unavailable",
+)
+@pytest.mark.parametrize("mode", FORWARD_FNS.keys())
+def test_flush_at_backward_completion(mode):
+    """The update runs after the complete top-level backward."""
+    model = _make_model()
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+        _train_step(model, x, FORWARD_FNS[mode], recipe)
+        assert counter.backward == 1
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+def test_backward_quantization_update_scope():
+    """Independent autograd calls in one logical backward update once."""
+    model = _make_model()
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        with te.backward_quantization_update_scope():
+            for _ in range(STEPS):
+                x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+                _train_step(model, x, _forward_plain, recipe)
+                assert counter.backward == 0
+        assert counter.backward == 1
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+def test_backward_quantization_update_scope_covers_delayed_wgrad():
+    """The update runs after delayed weight-gradient computation."""
+    model = te.Linear(HIDDEN, HIDDEN, bias=True, delay_wgrad_compute=True).cuda()
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        with te.backward_quantization_update_scope():
+            x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+            with te.autocast(enabled=True, recipe=recipe):
+                out = model(x)
+            out.float().sum().backward()
+            assert counter.backward == 0
+            model.backward_dw()
+            assert model.weight.grad is not None
+            assert counter.backward == 0
+        assert counter.backward == 1
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.parametrize("checkpoint_first_branch", [True, False])
+def test_branched_graph(checkpoint_first_branch):
+    """Two checkpointed sibling branches merging into one loss."""
+    branch_a = _make_model(num_layers=2, seed=1)
+    branch_b = _make_model(num_layers=2, seed=2)
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        for step in range(STEPS):
+            x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+            with te.autocast(enabled=True, recipe=recipe):
+                if checkpoint_first_branch:
+                    ya = te_checkpoint(_run_layers, branch_a, x, use_reentrant=True)
+                else:
+                    ya = _run_layers(branch_a, x)
+                yb = te_checkpoint(_run_layers, branch_b, x, use_reentrant=True)
+                out = ya + yb
+            loss = out.float().sum()
+            loss.backward()
+            FP8GlobalStateManager.flush_backward_quantization_update()
+            assert counter.forward == step + 1
+            assert counter.backward == step + 1
+            assert x.grad is not None and torch.isfinite(x.grad).all()
+            x.grad = None
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+def test_checkpointed_branch_without_backward():
+    """A checkpointed branch whose output never receives a gradient must not
+    strand the backward update (regression test for first-module ownership)."""
+    used = _make_model(num_layers=2, seed=1)
+    unused = _make_model(num_layers=2, seed=2)
+    recipe = DelayedScaling()
+
+    with _UpdateCounter() as counter:
+        for step in range(STEPS):
+            x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+            with te.autocast(enabled=True, recipe=recipe):
+                # The unused branch runs first, so any "first module owns the
+                # backward update" scheme would assign ownership to a frame
+                # whose backward never executes.
+                y_unused = te_checkpoint(_run_layers, unused, x, use_reentrant=True)
+                y = _run_layers(used, x)
+            loss = y.float().sum()
+            loss.backward()
+            del y_unused
+            FP8GlobalStateManager.flush_backward_quantization_update()
+            assert counter.backward == step + 1, "backward update was lost"
+            x.grad = None
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.parametrize("amax_compute_algo", ["max", "most_recent"])
+@pytest.mark.parametrize("mode", ["reentrant", "non_reentrant", "per_layer_reentrant", "nested"])
+def test_checkpointing_matches_plain_numerics(mode, amax_compute_algo):
+    """Delayed-scaling state must evolve identically with and without
+    activation checkpointing (duplicate updates would advance it faster)."""
+    forward_fn = FORWARD_FNS[mode]
+    recipe = DelayedScaling(amax_history_len=4, amax_compute_algo=amax_compute_algo)
+
+    def train(forward):
+        model = _make_model(seed=99)
+        losses = []
+        torch.manual_seed(777)
+        for _ in range(STEPS + 1):
+            x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+            losses.append(_train_step(model, x, forward, recipe).detach())
+        FP8GlobalStateManager.flush_backward_quantization_update()
+        state = []
+        for layer in model:
+            for key in ("scaling_fwd", "scaling_bwd"):
+                meta = layer.fp8_meta[key]
+                state.append((meta.scale.clone(), meta.amax_history.clone()))
+        return losses, state
+
+    losses_ref, state_ref = train(_forward_plain)
+    losses_ckpt, state_ckpt = train(forward_fn)
+
+    for step, (l_ref, l_ckpt) in enumerate(zip(losses_ref, losses_ckpt)):
+        torch.testing.assert_close(l_ckpt, l_ref, rtol=0, atol=0, msg=f"loss diverged @ {step}")
+    for (scale_ref, hist_ref), (scale_ckpt, hist_ckpt) in zip(state_ref, state_ckpt):
+        assert torch.equal(scale_ckpt, scale_ref), "scale diverged under checkpointing"
+        assert torch.equal(hist_ckpt, hist_ref), "amax history diverged under checkpointing"
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+@pytest.mark.parametrize("use_reentrant", [True, False])
+def test_checkpointed_eval_module(use_reentrant):
+    """Eval mode does not disable activation recompute or autograd."""
+    model = _make_model()
+    model.eval()
+    recipe = DelayedScaling()
+    x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+
+    with te.autocast(enabled=True, recipe=recipe):
+        out = te_checkpoint(_run_layers, model, x, use_reentrant=use_reentrant)
+    out.float().sum().backward()
+    FP8GlobalStateManager.flush_backward_quantization_update()
+
+    assert x.grad is not None and torch.isfinite(x.grad).all()
+
+
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+def test_exception_in_checkpointed_forward():
+    """A failing checkpointed forward must not corrupt update scheduling."""
+    model = _make_model()
+    recipe = DelayedScaling()
+
+    def failing(model, x):
+        def fn(x):
+            model[0](x)
+            raise RuntimeError("boom")
+
+        return te_checkpoint(fn, x, use_reentrant=True)
+
+    with _UpdateCounter() as counter:
+        x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+        with pytest.raises(RuntimeError, match="boom"):
+            _train_step(model, x, failing, recipe)
+
+        x = torch.randn(BATCH, HIDDEN, device="cuda", requires_grad=True)
+        _train_step(model, x, _forward_reentrant, recipe)
+        FP8GlobalStateManager.flush_backward_quantization_update()
+        assert counter.backward == 1
+        assert torch.isfinite(x.grad).all()
diff --git a/transformer_engine/pytorch/__init__.py b/transformer_engine/pytorch/__init__.py
index 2b1803bfb2..26f10bef9f 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 backward_quantization_update_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..4eaf640a20 100644
--- a/transformer_engine/pytorch/distributed.py
+++ b/transformer_engine/pytorch/distributed.py
@@ -62,9 +62,6 @@
 
 _USE_REENTRANT_ACTIVATION_RECOMPUTE = True
 
-_IN_ACTIVATION_RECOMPUTE_REGION = False
-_ACTIVATION_RECOMPUTE_PHASE = False
-
 
 _ALL_ACTIVE_RNG_STATES = {}
 
@@ -247,43 +244,52 @@ 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
         self.recompute_phase = recompute_phase
 
     def __enter__(self):
-        global _IN_ACTIVATION_RECOMPUTE_REGION, _ACTIVATION_RECOMPUTE_PHASE
+        qstate = FP8GlobalStateManager.quantization_state
+        self._prev_region = qstate.in_activation_recompute_region
+        self._prev_phase = qstate.activation_recompute_phase
+        self._prev_forward_phase = qstate.activation_recompute_forward_phase
         # Track the checkpoint region independently of the FP8 state at entry.
         # A checkpointed callable may open its own FP8 autocast context (for
         # example, to select precision per layer). Delayed-scaling modules in
         # that inner context must still save their scale and amax metadata for
         # the recompute forward.
-        _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)
+        qstate.in_activation_recompute_region = self._prev_region or self.activation_recompute
+        qstate.activation_recompute_phase = self._prev_phase or (
+            self.activation_recompute and self.recompute_phase
+        )
+        qstate.activation_recompute_forward_phase = self._prev_forward_phase or (
+            self.activation_recompute and not self.recompute_phase
+        )
 
     def __exit__(self, *exc_details):
-        global _IN_ACTIVATION_RECOMPUTE_REGION, _ACTIVATION_RECOMPUTE_PHASE
-        _IN_ACTIVATION_RECOMPUTE_REGION = False
-        _ACTIVATION_RECOMPUTE_PHASE = False
+        qstate = FP8GlobalStateManager.quantization_state
+        qstate.in_activation_recompute_region = self._prev_region
+        qstate.activation_recompute_phase = self._prev_phase
+        qstate.activation_recompute_forward_phase = self._prev_forward_phase
 
 
 def is_fp8_activation_recompute_enabled() -> bool:
     """Whether we are in an activation recompute region with FP8 currently enabled"""
-    return _IN_ACTIVATION_RECOMPUTE_REGION and FP8GlobalStateManager.is_fp8_enabled()
+    return (
+        FP8GlobalStateManager.quantization_state.in_activation_recompute_region
+        and FP8GlobalStateManager.is_fp8_enabled()
+    )
 
 
 def in_fp8_activation_recompute_phase() -> bool:
     """Return global boolean"""
-    return _ACTIVATION_RECOMPUTE_PHASE
+    return FP8GlobalStateManager.quantization_state.activation_recompute_phase
+
+
+def in_fp8_activation_recompute_forward_phase() -> bool:
+    """Whether an activation-checkpoint forward frame is active."""
+    return FP8GlobalStateManager.quantization_state.activation_recompute_forward_phase
 
 
 def _get_active_autocast_contexts():
diff --git a/transformer_engine/pytorch/module/base.py b/transformer_engine/pytorch/module/base.py
index e9a65c3648..027ea1cd04 100644
--- a/transformer_engine/pytorch/module/base.py
+++ b/transformer_engine/pytorch/module/base.py
@@ -42,6 +42,7 @@
 from ..distributed import (
     gather_along_first_dim,
     is_fp8_activation_recompute_enabled,
+    in_fp8_activation_recompute_forward_phase,
     in_fp8_activation_recompute_phase,
     _fsdp_gather_tensors,
 )
@@ -1631,9 +1632,12 @@ def prepare_forward(
                 if not FP8GlobalStateManager.fp8_graph_capturing():
                     FP8GlobalStateManager.add_fp8_tensors_to_global_buffer(self.fp8_meta)
 
-                # Activation recomputation is used and this is the first forward phase.
-                if self.training and is_fp8_activation_recompute_enabled():
-                    FP8GlobalStateManager.copy_forward_fp8_meta_tensors_for_recompute(self.fp8_meta)
+        if (
+            delayed_scaling_recipe
+            and is_fp8_activation_recompute_enabled()
+            and in_fp8_activation_recompute_forward_phase()
+        ):
+            FP8GlobalStateManager.copy_forward_fp8_meta_tensors_for_recompute(self.fp8_meta)
 
         nvtx_range_push(self.__class__.__name__ + " forward")
         if not allow_non_contiguous and not inp.is_contiguous():
diff --git a/transformer_engine/pytorch/module/grouped_linear.py b/transformer_engine/pytorch/module/grouped_linear.py
index fe283cf1a3..a9599f039c 100644
--- a/transformer_engine/pytorch/module/grouped_linear.py
+++ b/transformer_engine/pytorch/module/grouped_linear.py
@@ -49,6 +49,7 @@
     is_fp8_activation_recompute_enabled,
     in_fp8_activation_recompute_phase,
 )
+from ..graph import is_graph_capturing
 from ..distributed_weight import (
     is_distributed_weight,
     materialize_weight_for_forward,
@@ -962,12 +963,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.should_request_backward_quantization_update = ctx.fp8 and requires_grad(
+                inp, weights[0], biases[0]
+            )
             ctx.wgrad_store = wgrad_store
             ctx.debug = False
             ctx.save_original_input = save_original_input
@@ -1341,12 +1339,9 @@ 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.should_request_backward_quantization_update = ctx.fp8 and requires_grad(
+                inp, weights[0], biases[0]
+            )
             ctx.wgrad_store = wgrad_store
             ctx.debug = debug
             ctx.save_original_input = save_original_input
@@ -1364,7 +1359,7 @@ 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.should_request_backward_quantization_update = False
 
         # [*, in_features] -> [*, out_features] except first dimension changes for SP
         return out.view(-1, *inp.shape[1:-1], out.shape[-1]), new_workspaces
@@ -1616,8 +1611,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.should_request_backward_quantization_update and not is_graph_capturing():
+            FP8GlobalStateManager.request_backward_quantization_update()
         return (
             dgrad.view(ctx.inp_shape) if ctx.requires_dgrad else None,
             None,  # m_splits
@@ -1880,8 +1875,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.should_request_backward_quantization_update and not is_graph_capturing():
+            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..f327fd7a6b 100644
--- a/transformer_engine/pytorch/module/layernorm_linear.py
+++ b/transformer_engine/pytorch/module/layernorm_linear.py
@@ -52,7 +52,6 @@
     symmetric_all_reduce,
     reduce_scatter_along_first_dim,
     gather_along_first_dim,
-    in_fp8_activation_recompute_phase,
     _fsdp_scatter_tensors,
     _fsdp_gather_tensors,
 )
@@ -588,13 +587,9 @@ 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.should_request_backward_quantization_update = ctx.fp8 and requires_grad(
+                inp, ln_weight, ln_bias, weight, bias
+            )
             ctx.wgrad_store = wgrad_store
             ctx.debug = debug
 
@@ -609,7 +604,7 @@ 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.should_request_backward_quantization_update = False
 
         # ------------------------------------------------------
         # Cached state for backward pass is ready...
@@ -1184,10 +1179,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.should_request_backward_quantization_update and not is_graph_capturing():
+            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..c5ffbe3727 100644
--- a/transformer_engine/pytorch/module/layernorm_mlp.py
+++ b/transformer_engine/pytorch/module/layernorm_mlp.py
@@ -58,7 +58,6 @@
     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,
@@ -908,15 +907,9 @@ 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(
+            ctx.should_request_backward_quantization_update = 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
             if is_recomputation:  # return the recomputed tensors
@@ -1806,8 +1799,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.should_request_backward_quantization_update and not is_graph_capturing():
+            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..6e21c219ce 100644
--- a/transformer_engine/pytorch/module/linear.py
+++ b/transformer_engine/pytorch/module/linear.py
@@ -233,8 +233,8 @@ 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
+    # --- Quantization state update bookkeeping ---
+    should_request_backward_quantization_update: bool = False
 
     # --- Misc ---
     cpu_offloading: bool = False
@@ -255,16 +255,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]]:
@@ -1439,9 +1429,9 @@ def forward(
                 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()
+                bwd_args.should_request_backward_quantization_update = True
             if fwd_args.backward_override is not None:
-                bwd_args.reduce_and_update_bwd_fp8_tensors = False
+                bwd_args.should_request_backward_quantization_update = False
 
         return out, new_weight_workspace
 
@@ -1459,15 +1449,15 @@ 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
+        should_request_backward_quantization_update = (
+            bwd_args.should_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 should_request_backward_quantization_update and not is_graph_capturing():
+            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..292c6f0d9f 100644
--- a/transformer_engine/pytorch/ops/fuser.py
+++ b/transformer_engine/pytorch/ops/fuser.py
@@ -227,10 +227,11 @@ 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()
+            # Whether to request a recipe update in backward pass
+            should_request_backward_quantization_update = (
+                fuser.first_op_requiring_backward < fuser._num_basic_ops
+                and FP8GlobalStateManager.is_fp8_enabled()
+            )
 
             # Other context
             func_ctx.backward_ops = fuser._backward_ops
@@ -243,7 +244,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.should_request_backward_quantization_update = (
+                should_request_backward_quantization_update
+            )
 
         # Mark output tensors as not deletable in backward
         for tensor in itertools.chain(
@@ -384,8 +387,8 @@ def backward(
         ]
 
         # 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.should_request_backward_quantization_update and not _is_graph_capturing():
+            FP8GlobalStateManager.request_backward_quantization_update()
 
         return (
             dx,  # input_
diff --git a/transformer_engine/pytorch/quantization.py b/transformer_engine/pytorch/quantization.py
index 98c67be922..4c12389ca0 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",
+    "backward_quantization_update_scope",
     "quantized_model_init",
     "is_fp8_available",
     "is_mxfp8_available",
@@ -400,6 +401,12 @@ 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_queued: bool = False
+    backward_quantization_update_scope_depth: int = 0
+    in_activation_recompute_region: bool = False
+    activation_recompute_phase: bool = False
+    activation_recompute_forward_phase: bool = False
     fp8_graph_capturing: bool = False
     autocast_depth: int = 0
     global_amax_buffer: Dict[str, list] = field(default_factory=dict)
@@ -653,62 +660,98 @@ 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:
-        """Delayed scaling only. Concatenate, reduce, and split amaxes in the global buffer."""
-        # global_amax_buffer should only be non-empty for fp8 delayed scaling
+        """Run the per-recipe quantization state updates over the global buffers.
+
+        The manager only orchestrates *when* updates run (forward at autocast
+        exit, backward after the backward pass); *how* a recipe updates its
+        state is defined by `RecipeState.reduce_and_update_global_state`
+        overrides. Currently only delayed scaling has one, so the buffers
+        are empty for other recipes.
+        """
         qstate = cls.quantization_state
         for (
             buffer_key,
             amax_buffer,
         ) in qstate.global_amax_buffer.items():
-            # Check for forward or backward reduction.
+            # Check for forward or backward update.
             fwd_update, autocast_key = cls.split_key_in_buffer(buffer_key)
             if fwd_update != forward:
                 continue
             if len(amax_buffer) == 0:
                 continue
 
-            # Retrieve autocast specific args and concat amaxes.
             recipe, group = qstate.autocast_arguments[autocast_key]
-            contiguous_amax = torch.cat(amax_buffer)
+            RecipeState.class_for_recipe(recipe).reduce_and_update_global_state(
+                recipe,
+                group,
+                forward,
+                amax_buffer,
+                qstate.global_amax_history_buffer[buffer_key],
+                qstate.global_scale_buffer[buffer_key],
+            )
 
-            # Reduction.
-            if (
-                recipe.reduce_amax
-                and torch.distributed.is_initialized()
-                and torch.distributed.get_world_size(group=group) > 1
-            ):
-                cls.reduce_tensor_across_group_op_max(contiguous_amax, group)
+    # Compatibility alias (used by Megatron-Core).
+    reduce_and_update_fp8_tensors = reduce_and_update_quantization_state
 
-            # Amax and scale update.
-            unfused_update = (
-                bool(int(os.getenv("NVTE_UNFUSED_FP8_UPDATE", "0")))
-                or callable(recipe.amax_compute_algo)
-                or callable(recipe.scaling_factor_compute_algo)
-            )
+    @classmethod
+    def request_backward_quantization_update(cls) -> None:
+        """Request a backward quantization state update (e.g. delayed scaling
+        amax reduction) after the enclosing logical backward."""
+        qstate = cls.quantization_state
+        qstate.pending_backward_quantization_update = True
+        cls._queue_backward_quantization_update_callback()
 
-            if not unfused_update:
-                tex.fused_amax_and_scale_update_after_reduction(
-                    contiguous_amax,
-                    qstate.global_amax_history_buffer[buffer_key],
-                    qstate.global_scale_buffer[buffer_key],
-                    recipe.amax_compute_algo,
-                    get_fp8_te_dtype(recipe, forward),
-                    recipe.margin,
-                )
-            else:
-                split_and_copy(contiguous_amax, amax_buffer, [x.numel() for x in amax_buffer])
-
-                for amax_history, scale in zip(
-                    qstate.global_amax_history_buffer[buffer_key],
-                    qstate.global_scale_buffer[buffer_key],
-                ):
-                    _amax_and_scale_update(
-                        amax_history, scale, get_fp8_max(recipe, forward), recipe
-                    )
+    @classmethod
+    def _queue_backward_quantization_update_callback(cls) -> None:
+        """Queue the pending update after the top-level autograd task."""
+        qstate = cls.quantization_state
+        if (
+            qstate.backward_quantization_update_scope_depth > 0
+            or qstate.backward_quantization_update_callback_queued
+            or torch.compiler.is_compiling()
+            or cls.fp8_graph_capturing()
+            or torch._C._current_graph_task_id() == -1
+        ):
+            return
+        queue_callback = getattr(torch.autograd.graph, "queue_callback", None)
+        if queue_callback is None:
+            return
+        qstate.backward_quantization_update_callback_queued = True
+        try:
+            queue_callback(cls._run_backward_quantization_update_callback)
+        except RuntimeError:
+            qstate.backward_quantization_update_callback_queued = False
+            raise
+
+    @classmethod
+    def _run_backward_quantization_update_callback(cls) -> None:
+        """Flush a pending update at the end of the enclosing backward."""
+        qstate = cls.quantization_state
+        qstate.backward_quantization_update_callback_queued = False
+        if qstate.backward_quantization_update_scope_depth == 0:
+            cls.flush_backward_quantization_update()
+
+    @classmethod
+    def flush_backward_quantization_update(cls) -> None:
+        """Run the pending backward quantization state 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(
@@ -740,6 +783,23 @@ def autocast_enter(
         fp8_recipe = get_default_fp8_recipe() if fp8_recipe is None else fp8_recipe
         autocast_key = cls.get_unique_autocast_key(fp8_recipe, fp8_group)
         qstate = cls.quantization_state
+
+        # Fallback for runtimes without the autograd callback API. Never flush
+        # during activation recompute, an explicit backward update scope, or a
+        # running backward pass.
+        if (
+            qstate.autocast_depth == 0
+            and qstate.pending_backward_quantization_update
+            and qstate.backward_quantization_update_scope_depth == 0
+        ):
+            if (
+                not _graph
+                and not qstate.activation_recompute_phase
+                and not torch.compiler.is_compiling()
+                and not cls.fp8_graph_capturing()
+                and torch._C._current_graph_task_id() == -1
+            ):
+                cls.flush_backward_quantization_update()
         qstate.autocast_arguments[autocast_key] = (
             fp8_recipe,
             fp8_group,
@@ -776,10 +836,18 @@ def autocast_exit(cls, enabled: bool, _graph: bool) -> None:
         # Reduce only the non-FP8 weight modules here.
         # FP8 weight modules are reduced at the end of the optimizer
         # step after the weight amax is populated.
-        if enabled and qstate.autocast_depth == 0 and not _graph and torch.is_grad_enabled():
+        # Skip during activation recompute: the recompute re-enters an autocast, and
+        # updating here a second time would advance delayed-scaling state twice per step.
+        if (
+            enabled
+            and qstate.autocast_depth == 0
+            and not _graph
+            and not qstate.activation_recompute_phase
+            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:
@@ -799,15 +867,12 @@ def copy_forward_fp8_meta_tensors_for_recompute(cls, fp8_meta: Dict[str, Any]) -
         ]
 
         qstate = cls.quantization_state
-        if buffer_position_key in fp8_meta:
-            qstate.fp8_tensors_recompute_buffer[fp8_meta[buffer_position_key]].append(to_copy)
-        else:
-            if len(qstate.fp8_tensors_recompute_buffer) == 0:
-                qstate.fp8_tensors_recompute_buffer = [deque()]
-            else:
-                qstate.fp8_tensors_recompute_buffer.append(deque())
-            qstate.fp8_tensors_recompute_buffer[-1].append(to_copy)
-            fp8_meta[buffer_position_key] = len(qstate.fp8_tensors_recompute_buffer) - 1
+        buffer_position = fp8_meta.get(buffer_position_key)
+        if buffer_position is None or buffer_position >= len(qstate.fp8_tensors_recompute_buffer):
+            qstate.fp8_tensors_recompute_buffer.append(deque())
+            buffer_position = len(qstate.fp8_tensors_recompute_buffer) - 1
+            fp8_meta[buffer_position_key] = buffer_position
+        qstate.fp8_tensors_recompute_buffer[buffer_position].append(to_copy)
 
     @classmethod
     def get_old_fp8_meta_tensors_for_recompute(cls, fp8_meta: Dict[str, Any]) -> None:
@@ -818,16 +883,22 @@ def get_old_fp8_meta_tensors_for_recompute(cls, fp8_meta: Dict[str, Any]) -> Non
         if not _has_delayed_scaling_state(fp8_meta):
             return
 
+        # Retrieve stashed amaxes and scales from phase 1 pre forward.
+        buffer_position_key = "global_fp8_buffer_pos_fwd_recompute"
+        buffer_position = fp8_meta.get(buffer_position_key)
+        buffers = cls.quantization_state.fp8_tensors_recompute_buffer
+        if (
+            buffer_position is None
+            or buffer_position >= len(buffers)
+            or not buffers[buffer_position]
+        ):
+            raise RuntimeError("FP8 activation recompute metadata stash is missing")
+        stashed_fp8_meta = buffers[buffer_position].popleft()
+
         # Store updated amaxes and scales from phase 1 post forward.
         fp8_meta["updated_amax_history_fwd"] = fp8_meta["scaling_fwd"].amax_history.clone()
         fp8_meta["updated_scale_fwd"] = fp8_meta["scaling_fwd"].scale.clone()
 
-        # Retrieve stashed amaxes and scales from phase 1 pre forward.
-        buffer_position_key = "global_fp8_buffer_pos_fwd_recompute"
-        stashed_fp8_meta = cls.quantization_state.fp8_tensors_recompute_buffer[
-            fp8_meta[buffer_position_key]
-        ].popleft()
-
         # Replace amaxes and scales with stashed values for phase 2 forward
         fp8_meta["scaling_fwd"].amax_history.copy_(stashed_fp8_meta[0])
         fp8_meta["scaling_fwd"].scale.copy_(stashed_fp8_meta[1])
@@ -843,6 +914,28 @@ def restore_fp8_meta_tensors(fp8_meta: Dict[str, Any]) -> None:
         fp8_meta["scaling_fwd"].scale.copy_(fp8_meta["updated_scale_fwd"])
 
 
+@contextmanager
+def backward_quantization_update_scope() -> None:
+    """Group quantization state updates from multiple backward calls.
+
+    Use this around one logical backward that is implemented with multiple
+    autograd backward calls or schedules gradient work after autograd returns.
+    The scope must include delayed work such as ``module.backward_dw()``.
+    Nested scopes flush once when the outermost scope exits.
+    """
+    qstate = FP8GlobalStateManager.quantization_state
+    qstate.backward_quantization_update_scope_depth += 1
+    try:
+        yield
+    finally:
+        qstate.backward_quantization_update_scope_depth -= 1
+        if qstate.backward_quantization_update_scope_depth == 0:
+            if torch._C._current_graph_task_id() == -1:
+                FP8GlobalStateManager.flush_backward_quantization_update()
+            else:
+                FP8GlobalStateManager._queue_backward_quantization_update_callback()
+
+
 @contextmanager
 def fp8_model_init(
     enabled: bool = True,
@@ -1323,21 +1416,7 @@ def create(
 
         """
 
-        cls = None
-        if recipe.delayed():
-            cls = DelayedScalingRecipeState
-        elif recipe.mxfp8():
-            cls = MXFP8BlockScalingRecipeState
-        elif recipe.float8_current_scaling():
-            cls = Float8CurrentScalingRecipeState
-        elif recipe.float8_block_scaling():
-            cls = Float8BlockScalingRecipeState
-        elif recipe.nvfp4():
-            cls = NVFP4BlockScalingRecipeState
-        elif recipe.custom():
-            cls = CustomRecipeState
-        else:
-            raise ValueError(f"{recipe.__class__.__name__} is not supported")
+        cls = RecipeState.class_for_recipe(recipe)
         return cls(
             recipe,
             mode=mode,
@@ -1346,6 +1425,42 @@ def create(
             roles=roles,
         )
 
+    @staticmethod
+    def class_for_recipe(recipe: Recipe) -> type:
+        """Resolve the RecipeState subclass handling a recipe."""
+        if recipe.delayed():
+            return DelayedScalingRecipeState
+        if recipe.mxfp8():
+            return MXFP8BlockScalingRecipeState
+        if recipe.float8_current_scaling():
+            return Float8CurrentScalingRecipeState
+        if recipe.float8_block_scaling():
+            return Float8BlockScalingRecipeState
+        if recipe.nvfp4():
+            return NVFP4BlockScalingRecipeState
+        if recipe.custom():
+            return CustomRecipeState
+        raise ValueError(f"{recipe.__class__.__name__} is not supported")
+
+    @classmethod
+    def reduce_and_update_global_state(
+        cls,
+        recipe: Recipe,
+        group: Optional[dist_group_type],
+        forward: bool,
+        amax_buffer: List[torch.Tensor],
+        amax_history_buffer: List[torch.Tensor],
+        scale_buffer: List[torch.Tensor],
+    ) -> None:
+        """Run this recipe's post-step update of process-global state.
+
+        Called by `FP8GlobalStateManager` once per autocast key, with the
+        state buffers of all participating modules — after the forward pass
+        (autocast exit) and after the backward pass. The manager decides
+        *when* updates run; recipes define *what* they update. The default
+        is a no-op: most recipes keep no process-global state.
+        """
+
     @abc.abstractmethod
     def make_quantizers(self) -> list:
         """Convert recipe state to quantizers.
@@ -1414,6 +1529,47 @@ class DelayedScalingRecipeState(RecipeState):
     # See ``RecipeState.inherit_state_from``.
     _persistent_state_buffers = ("scale", "amax_history")
 
+    @classmethod
+    def reduce_and_update_global_state(
+        cls,
+        recipe: DelayedScaling,
+        group: Optional[dist_group_type],
+        forward: bool,
+        amax_buffer: List[torch.Tensor],
+        amax_history_buffer: List[torch.Tensor],
+        scale_buffer: List[torch.Tensor],
+    ) -> None:
+        """Concatenate, reduce, and split amaxes, then recompute scales."""
+        contiguous_amax = torch.cat(amax_buffer)
+
+        if (
+            recipe.reduce_amax
+            and torch.distributed.is_initialized()
+            and torch.distributed.get_world_size(group=group) > 1
+        ):
+            FP8GlobalStateManager.reduce_tensor_across_group_op_max(contiguous_amax, group)
+
+        unfused_update = (
+            bool(int(os.getenv("NVTE_UNFUSED_FP8_UPDATE", "0")))
+            or callable(recipe.amax_compute_algo)
+            or callable(recipe.scaling_factor_compute_algo)
+        )
+
+        if not unfused_update:
+            tex.fused_amax_and_scale_update_after_reduction(
+                contiguous_amax,
+                amax_history_buffer,
+                scale_buffer,
+                recipe.amax_compute_algo,
+                get_fp8_te_dtype(recipe, forward),
+                recipe.margin,
+            )
+        else:
+            split_and_copy(contiguous_amax, amax_buffer, [x.numel() for x in amax_buffer])
+
+            for amax_history, scale in zip(amax_history_buffer, scale_buffer):
+                _amax_and_scale_update(amax_history, scale, get_fp8_max(recipe, forward), recipe)
+
     def __init__(
         self,
         recipe: DelayedScaling,