diff --git a/tests/pytorch/test_torch_compile.py b/tests/pytorch/test_torch_compile.py
index e1c6eaf599..f8f5c4de08 100644
--- a/tests/pytorch/test_torch_compile.py
+++ b/tests/pytorch/test_torch_compile.py
@@ -4,10 +4,12 @@
 
 import abc
 import contextlib
+import dataclasses
 import os
 import re
 import sys
 import warnings
+from typing import Union
 
 import pytest
 import torch
@@ -37,11 +39,17 @@
 from transformer_engine.pytorch.module.base import TransformerEngineBaseModule
 from transformer_engine.pytorch.quantization import FP8GlobalStateManager, QuantizerRole
 from transformer_engine.pytorch.ops.basic.basic_linear import BasicLinear
+from transformer_engine.pytorch.ops.fuser import OperationFuser
+from transformer_engine.pytorch.ops.op import BasicOperation, OperationContext
 from transformer_engine.pytorch.tensor.float8_tensor import Float8CurrentScalingQuantizer
 from transformer_engine.pytorch.tensor.float8_blockwise_tensor import Float8BlockQuantizer
 from transformer_engine.pytorch.tensor.mxfp8_tensor import MXFP8Quantizer
 from transformer_engine.pytorch.tensor.nvfp4_tensor import NVFP4Quantizer
-from transformer_engine.pytorch.quantized_tensor import QuantizedTensor, Quantizer
+from transformer_engine.pytorch.quantized_tensor import (
+    QuantizedTensor,
+    QuantizedTensorStorage,
+    Quantizer,
+)
 from transformer_engine.pytorch.dynamo import TensorSpec, to_tensor_spec
 from transformer_engine.pytorch import (
     is_fp8_available,
@@ -2339,3 +2347,302 @@ def fn(inp):
             "Unexpected recompilation(s) across different batch sizes: "
             f"{unique_graphs_after - unique_graphs_baseline} extra graph(s) compiled"
         )
+
+
+# --------------------------------------------------------------------------- #
+# transformer_engine.pytorch.ops under torch.compile
+# --------------------------------------------------------------------------- #
+
+
+@dataclasses.dataclass(slots=True)
+class _AffineFwdArgs:
+    input_: torch.Tensor
+    weight: torch.Tensor
+    gain: float = 1.0
+    offset: Union[torch.Tensor, QuantizedTensorStorage] = None
+
+
+@dataclasses.dataclass(slots=True)
+class _AffineBwdArgs:
+    grad_output: torch.Tensor
+    input_: torch.Tensor
+    weight: torch.Tensor
+    gain: float
+
+
+class _AffineOp(BasicOperation):
+    """Learnable scale with read-only gain and offset kwargs."""
+
+    fwd_args_type = _AffineFwdArgs
+    bwd_args_type = _AffineBwdArgs
+
+    def __init__(self, weight=2.0, dtype=torch.float32):
+        super().__init__()
+        self.weight = torch.nn.Parameter(torch.tensor(weight, dtype=dtype, device="cuda"))
+
+    @classmethod
+    def forward_compute(cls, args):
+        output = args.input_ * args.weight * args.gain
+        offset = args.offset
+        if isinstance(offset, QuantizedTensorStorage):
+            offset = offset.dequantize()
+        if offset is not None:
+            output = output + offset
+        return output, [()], ()
+
+    @classmethod
+    def forward_fake(cls, args):
+        return args.input_, [()], ()
+
+    @classmethod
+    def backward_compute(cls, args):
+        dy = args.grad_output
+        return dy * args.weight * args.gain, [((dy * args.input_).sum() * args.gain,)], [()]
+
+    @classmethod
+    def backward_fake(cls, args):
+        dy = args.grad_output
+        return dy, [(TensorSpec(shape=(), dtype=dy.dtype, device=dy.device),)], [()]
+
+    def pack_forward_args(self, basic_op_ctxs, input_, *, basic_op_kwargs, **unused):
+        return _AffineFwdArgs(input_, self.weight, **basic_op_kwargs[0])
+
+    def forward_setup_context(self, basic_op_ctxs, args, aux):
+        ctx = basic_op_ctxs[0]
+        ctx.save_for_backward(args.input_, args.weight)
+        ctx.gain = args.gain
+
+    def pack_backward_args(self, basic_op_ctxs, grad_output, **unused):
+        ctx = basic_op_ctxs[0]
+        return _AffineBwdArgs(grad_output, *ctx.saved_tensors, ctx.gain)
+
+
+class _BackwardAffinePair(te.ops.FusedOperation):
+    def fuser_backward(self, basic_op_ctxs, grad_output, **unused):
+        grads = []
+        for op, ctx in reversed(list(zip(self.basic_ops, basic_op_ctxs))):
+            grad_output, params, _ = op.fuser_backward(
+                [ctx], grad_output, basic_op_grad_extra_outputs=[()]
+            )
+            grads.insert(0, params[0])
+        return grad_output, grads, [(), ()]
+
+
+@dataclasses.dataclass(slots=True)
+class _AffinePairFwdArgs:
+    input_: torch.Tensor
+    weight0: torch.Tensor
+    weight1: torch.Tensor
+    residual: torch.Tensor
+
+
+@dataclasses.dataclass(slots=True)
+class _AffinePairBwdArgs:
+    grad_output: torch.Tensor
+    input_: torch.Tensor
+    intermediate: torch.Tensor
+    weight0: torch.Tensor
+    weight1: torch.Tensor
+    grad_extra_output: torch.Tensor
+
+
+class _AffinePair(te.ops.FusedOperation):
+    """Two scales with a residual input and a squared intermediate output."""
+
+    fwd_args_type = _AffinePairFwdArgs
+    bwd_args_type = _AffinePairBwdArgs
+
+    @classmethod
+    def forward_compute(cls, args):
+        intermediate = args.input_ * args.weight0
+        output = intermediate * args.weight1 + args.residual
+        return output, [(), (intermediate.square(), None)], (intermediate,)
+
+    @classmethod
+    def forward_fake(cls, args):
+        return args.input_, [(), (args.input_, None)], (args.input_,)
+
+    @classmethod
+    def backward_compute(cls, args):
+        dy = args.grad_output
+        du = dy * args.weight1 + 2 * args.intermediate * args.grad_extra_output
+        return (
+            du * args.weight0,
+            [((du * args.input_).sum(),), ((dy * args.intermediate).sum(),)],
+            [(), (dy.clone(),)],
+        )
+
+    @classmethod
+    def backward_fake(cls, args):
+        dy = args.grad_output
+        scalar = TensorSpec(shape=(), dtype=dy.dtype, device=dy.device)
+        return dy, [(scalar,), (scalar,)], [(), (dy,)]
+
+    def pack_forward_args(self, basic_op_ctxs, input_, *, basic_op_extra_inputs, **unused):
+        return _AffinePairFwdArgs(
+            input_,
+            self.basic_ops[0].weight,
+            self.basic_ops[1].weight,
+            basic_op_extra_inputs[1][0],
+        )
+
+    def forward_setup_context(self, basic_op_ctxs, args, aux):
+        basic_op_ctxs[0].save_for_backward(args.input_, args.weight0)
+        basic_op_ctxs[1].save_for_backward(aux[0], args.weight1)
+
+    def pack_backward_args(self, basic_op_ctxs, grad_output, *, basic_op_grad_extra_outputs):
+        x, weight0 = basic_op_ctxs[0].saved_tensors
+        intermediate, weight1 = basic_op_ctxs[1].saved_tensors
+        return _AffinePairBwdArgs(
+            grad_output, x, intermediate, weight0, weight1, basic_op_grad_extra_outputs[1][0]
+        )
+
+
+def _compile_with_graphs(fn):
+    graphs = []
+
+    def backend(graph, inputs):
+        graphs.append(graph)
+        return torch._dynamo.lookup_backend("inductor")(graph, inputs)
+
+    return torch.compile(fn, fullgraph=True, backend=backend), graphs
+
+
+def _assert_custom_ops(graphs, name, present=True):
+    targets = {
+        str(node.target).removesuffix(".default").removesuffix("_base")
+        for graph in graphs
+        for module in graph.modules()
+        if isinstance(module, torch.fx.GraphModule)
+        for node in module.graph.nodes
+        if node.op == "call_function"
+    }
+    for suffix in ("", "_backward"):
+        assert (f"transformer_engine_compile.{name}{suffix}" in targets) == present, targets
+
+
+def _check_ops(fn, model, x, dy, kwargs=None):
+    """Compare the pipeline with native PyTorch forward and autograd."""
+    params = tuple(model.parameters())
+    kwargs = kwargs or {}
+    offset = kwargs.get("offset", 0)
+    if isinstance(offset, QuantizedTensorStorage):
+        offset = offset.dequantize()
+    reference = x
+    for i, weight in enumerate(params):
+        reference = reference * weight
+        if i == 0:
+            reference = reference * kwargs.get("gain", 1.0) + offset
+    output = fn(x, op_kwargs={0: kwargs})
+    actual = output, torch.autograd.grad(output, (x, *params), dy)
+    expected = reference, torch.autograd.grad(reference, (x, *params), dy)
+    torch.testing.assert_close(actual, expected)
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+@pytest.mark.parametrize("dtype", [torch.float32, torch.bfloat16])
+@pytest.mark.parametrize(
+    "case,reason",
+    [
+        ("single", None),
+        ("multi", "several operations"),
+        ("backward_fusion", "backward fusion"),
+        ("legacy", "without a custom op"),
+    ],
+)
+def test_te_ops_pipeline(case, reason, dtype, monkeypatch):
+    torch._dynamo.reset()
+    if case == "backward_fusion":
+
+        def fuse(ops, **unused):
+            if len(ops) == 2 and all(isinstance(op, _AffineOp) for op in ops):
+                return [_BackwardAffinePair(ops)]
+            return ops
+
+        monkeypatch.setattr(OperationFuser, "backward_fusion_functions", [fuse])
+    ops = (
+        [te.ops.Identity()]
+        if case == "legacy"
+        else [_AffineOp(float(i + 2), dtype) for i in range(1 if case == "single" else 2)]
+    )
+    model = te.ops.Sequential(*ops)
+    compiled, graphs = _compile_with_graphs(model)
+    # Keep products exact in BF16 across eager and fused reductions.
+    x = (torch.randint(-8, 9, (8, 16), device="cuda").to(dtype) / 8).requires_grad_()
+    dy = torch.randint_like(x, -8, 9) / 8
+    with pytest.warns(UserWarning, match=reason) if reason else contextlib.nullcontext():
+        _check_ops(compiled, model, x, dy)
+    _check_ops(model, model, x, dy)
+    _assert_custom_ops(graphs, "_affineop", present=reason is None)
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+@pytest.mark.parametrize("use_custom_ops", [False, True])
+def test_te_ops_fused_compute_contract(use_custom_ops):
+    """Exercise the shared interface directly; pipeline fusion stays gated under compile."""
+    torch._dynamo.reset()
+    fused = _AffinePair([_AffineOp(2.0), _AffineOp(3.0)])
+    x = torch.randn(8, 16, device="cuda", requires_grad=True)
+    residual = torch.randn_like(x, requires_grad=True)
+    dy, dextra = torch.randn_like(x), torch.randn_like(x)
+
+    def run(input_, extra_input, grad_output, grad_extra_output):
+        ctxs = [OperationContext(), OperationContext()]
+        output, extras = fused.fuser_forward(
+            ctxs,
+            input_,
+            basic_op_extra_inputs=[(), (extra_input,)],
+            prev_op_grad_output_quantizer=None,
+            next_op_input_quantizer=None,
+            basic_op_kwargs=[{}, {}],
+            use_custom_ops=use_custom_ops,
+        )
+        for ctx in ctxs:
+            ctx.saved_tensors = ctx.to_save
+        grads = fused.fuser_backward(
+            ctxs,
+            grad_output,
+            basic_op_grad_extra_outputs=[(), (grad_extra_output, None)],
+            use_custom_ops=use_custom_ops,
+        )
+        return output, extras, grads
+
+    graphs = []
+    if use_custom_ops:
+        run, graphs = _compile_with_graphs(run)
+    with torch.no_grad():
+        actual = run(x, residual, dy, dextra)
+    weight0, weight1 = tuple(fused.parameters())
+    intermediate = x * weight0
+    output, extra = intermediate * weight1 + residual, intermediate.square()
+    dx, dw0, dw1, dr = torch.autograd.grad(
+        (output, extra), (x, weight0, weight1, residual), (dy, dextra)
+    )
+    expected = output, [(), (extra, None)], (dx, [(dw0,), (dw1,)], [(), (dr,)])
+    torch.testing.assert_close(actual, expected)
+    _assert_custom_ops(graphs, "_affinepair", present=use_custom_ops)
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8)
+def test_te_ops_forward_kwargs_compile():
+    torch._dynamo.reset()
+    model = te.ops.Sequential(_AffineOp())
+    compiled, graphs = _compile_with_graphs(model)
+    quantizer = Float8CurrentScalingQuantizer(
+        fp8_dtype=tex.DType.kFloat8E4M3, device=torch.device("cuda")
+    )
+    x = torch.randn(8, 16, device="cuda", requires_grad=True)
+    dy = torch.randn_like(x)
+    for i, value in enumerate((0.5, 0.5, 1.5, -0.5)):
+        offset = quantizer(torch.full((16,), value, device="cuda"))
+        _check_ops(compiled, model, x, dy, {"offset": offset})
+        if i == 1:
+            warmed_graph_count = len(graphs)
+        elif i > 1:
+            assert len(graphs) == warmed_graph_count, "Tensor kwarg triggered recompilation"
+    _assert_custom_ops(graphs, "_affineop")
+    with pytest.warns(UserWarning, match="non-tensor keyword arguments"):
+        for gain in (3.0, 5.0):
+            _check_ops(compiled, model, x, dy, {"gain": gain})
+    _assert_custom_ops(graphs[-1:], "_affineop", present=False)
diff --git a/transformer_engine/pytorch/dynamo/__init__.py b/transformer_engine/pytorch/dynamo/__init__.py
index e42eb8f9f6..042c209c1f 100644
--- a/transformer_engine/pytorch/dynamo/__init__.py
+++ b/transformer_engine/pytorch/dynamo/__init__.py
@@ -6,7 +6,11 @@
 
 from .quantizer_opaque import register_value_opaque_quantizer, is_value_opaque_quantizer
 from .tensor_spec import TensorSpec, to_tensor_spec
-from .custom_op import register_custom_op, TensorOrQuantized
+from .custom_op import (
+    register_custom_op,
+    register_custom_op_with_autograd,
+    TensorOrQuantized,
+)
 
 __all__ = [
     "register_value_opaque_quantizer",
@@ -14,5 +18,6 @@
     "TensorSpec",
     "to_tensor_spec",
     "register_custom_op",
+    "register_custom_op_with_autograd",
     "TensorOrQuantized",
 ]
diff --git a/transformer_engine/pytorch/dynamo/custom_op.py b/transformer_engine/pytorch/dynamo/custom_op.py
index f887a48d95..a87374095a 100644
--- a/transformer_engine/pytorch/dynamo/custom_op.py
+++ b/transformer_engine/pytorch/dynamo/custom_op.py
@@ -6,11 +6,16 @@
 
 Registers TE modules' eager forward/backward as ``torch.library`` custom ops so
 ``torch.compile(fullgraph=True)`` traces them as single graph nodes.
-``register_custom_op`` is the entry point; ``module/linear.py`` is the first user.
+``register_custom_op_with_autograd`` is the entry point for a module that
+wires autograd on the op itself (``module/linear.py`` is the first user);
+``register_custom_op`` registers one op without autograd; ``ops/fuser.py`` uses
+separate registrations for forward and backward and drives autograd itself.
 
 A TE forward/backward implementation takes one dataclass argument
 (``fwd_arg_type`` / ``bwd_arg_type``, e.g. ``LinearFwdArgs``) whose fields mix
 tensors, quantized tensors, quantizers, process groups and plain Python values.
+The autograd-free API preserves nested tensor results; the autograd-wired API
+keeps its saved-tensor and context-metadata contract.
 
 A ``torch.library`` custom op is narrower: it only accepts flat schema slots
 (tensors plus opaque objects) and returns a flat ``Tensor[]``.
@@ -41,9 +46,9 @@
     only when its value is trivial (``None`` / all-``None``) at call time.
 
 What runs where. Each op registers a data-free fake (``register_fake``) so it
-traces under ``torch.compile`` without allocating. ``register_custom_op`` returns
-``forward_fn`` -- the drop-in for the eager ``autograd.Function.apply``. A forward
-call through it:
+traces under ``torch.compile`` without allocating.
+``register_custom_op_with_autograd`` returns ``forward_fn`` -- the drop-in for
+the eager ``autograd.Function.apply``. A forward call through it:
 
   * runs the fake ``fwd_fake_impl`` on ``TensorSpec`` descriptors (data-free; see
     ``tensor_spec.py``) and parses its result into an ``_OutputPlan`` -- the
@@ -98,6 +103,7 @@
 import torch
 
 from torch._prims_common import make_contiguous_strides_for
+from torch.utils._pytree import tree_flatten, tree_unflatten
 
 from .tensor_spec import TensorSpec, to_tensor_spec
 from ..quantized_tensor import (
@@ -111,6 +117,7 @@
 
 _TE_OP_NAMESPACE = "transformer_engine_compile"
 
+
 # Annotation for an op arg field that may hold a plain tensor, a quantized
 # tensor subclass or a *bare* ``QuantizedTensorStorage`` (the internal-quantizer
 # optimization). Matched exactly by ``_TensorOrQuantizedAdapter``.
@@ -824,14 +831,16 @@ def _pack_fwd_result(result: Any) -> List[torch.Tensor]:
     return flat
 
 
-def _pack_bwd_result(grads: Any, num_grad_inputs: int, op_qualname: str) -> List[torch.Tensor]:
+def _pack_bwd_result(
+    grads: Any, num_grad_inputs: Optional[int], op_qualname: str
+) -> List[torch.Tensor]:
     """Pack a backward-impl return tuple into the op's ``Tensor[]`` payload.
 
-    Each grad occupies exactly one slot (validated against ``num_grad_inputs``);
-    a :class:`TensorSpec` grad is materialized into a single tensor.
+    Each grad occupies exactly one slot (validated against ``num_grad_inputs``
+    when given); a :class:`TensorSpec` grad is materialized into a single tensor.
     """
     grads = list(grads)
-    if len(grads) != num_grad_inputs:
+    if num_grad_inputs is not None and len(grads) != num_grad_inputs:
         raise RuntimeError(
             f"{op_qualname} expected bwd_impl to return {num_grad_inputs} grads "
             f"(one per input_tensors_for_grad entry), got {len(grads)}"
@@ -930,7 +939,7 @@ def _slice_user_grads(
 
 
 # --------------------------------------------------------------------------- #
-# Op registration
+# Op registration: base and wrapper ops, autograd wiring, one op
 # --------------------------------------------------------------------------- #
 
 
@@ -1166,7 +1175,124 @@ def _all_quantized_tensor_subclasses() -> List[type]:
     return found
 
 
+@dataclasses.dataclass(frozen=True)
+class _RegisteredOp:
+    """One registered custom op: its arg plan and the base / wrapper definitions."""
+
+    plan: _ArgPlan
+    base_def: Any
+    base_op: Any
+    wrapper_def: Any
+    wrapper_op: Any
+
+    def __call__(self, args: Any) -> List[torch.Tensor]:
+        """Pack the args dataclass into slots and call the wrapper op."""
+        kwargs = self.plan.pack(args)
+        return self.wrapper_op(*[kwargs[name] for name in self.plan.slot_names])
+
+
+def _register_op(
+    *,
+    name: str,
+    arg_type: type,
+    impl: Callable[[Any], Any],
+    fake_impl: Callable[[Any], Any],
+    pack_result: Callable[[Any], List[torch.Tensor]],
+) -> _RegisteredOp:
+    """Define one two-tier custom op: the base kernel, the wrapper op that lets
+    ``QuantizedTensor`` subclasses be inputs, and the passthrough registrations.
+    """
+    plan = _parse_arg_type(arg_type)
+    schema = f"{plan.schema_str} -> Tensor[]"
+    subclasses = _all_quantized_tensor_subclasses()
+    slot_offsets = plan.tensor_or_quantized_offsets()
+    namespace = getattr(torch.ops, _TE_OP_NAMESPACE)
+
+    for registered_name in (name, f"{name}_base"):
+        if hasattr(namespace, registered_name):
+            raise ValueError(f"Custom op '{_TE_OP_NAMESPACE}::{registered_name}' already exists")
+
+    base_def = _register_base_op(
+        op_name=f"{name}_base",
+        schema_str=schema,
+        plan=plan,
+        impl=impl,
+        fake_impl=fake_impl,
+        pack_result=pack_result,
+    )
+    base_op = getattr(namespace, f"{name}_base")
+    wrapper_def = _register_wrapper_op(
+        wrapper_op_name=name,
+        schema_str=schema,
+        base_op=base_op,
+        slot_offsets=slot_offsets,
+        subclasses=subclasses,
+    )
+    wrapper_op = getattr(namespace, name)
+
+    rule = _make_dispatch_rule(_make_slot_forwarder(base_op, slot_offsets, subclasses))
+    for sub in subclasses:
+        wrapper_def.register_torch_dispatch(sub, rule)
+    _quantized_tensor_passthrough_ops.update((base_op.default, wrapper_op.default))
+
+    return _RegisteredOp(
+        plan=plan,
+        base_def=base_def,
+        base_op=base_op,
+        wrapper_def=wrapper_def,
+        wrapper_op=wrapper_op,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Op registration: a single autograd-free op, and the autograd-wired variant
+# --------------------------------------------------------------------------- #
+
+
 def register_custom_op(
+    *,
+    op_name: str,
+    arg_type: type,
+    impl: Callable[[Any], Any],
+    fake_impl: Callable[[Any], Any],
+) -> Optional[Callable[[Any], Any]]:
+    """Register one custom op without autograd wiring.
+
+    arg_type is a dataclass defining the input schema. Results may be nested
+    tuples/lists of fresh tensors or None; fake_impl mirrors them with TensorSpec.
+    The returned callable takes an args instance and preserves the result structure.
+    Returns None for unsupported registration APIs so callers can fall back to eager.
+    Duplicate operator names raise ValueError.
+    """
+
+    def pack_result(result):
+        values, _ = tree_flatten(result)
+        return [tensor for value in values for tensor in _flatten_value(value)]
+
+    try:
+        op = _register_op(
+            name=op_name,
+            arg_type=arg_type,
+            impl=impl,
+            fake_impl=fake_impl,
+            pack_result=pack_result,
+        )
+    except (ImportError, AttributeError, RuntimeError, TypeError) as e:
+        record_compile_disabled(
+            f"could not register custom op '{op_name}' ({type(e).__name__}: {e})"
+        )
+        return None
+
+    def call(args):
+        spec_args = _spec_view(args, op.plan.tensor_field_names())
+        specs, structure = tree_flatten(fake_impl(spec_args))
+        out_plan = _OutputPlan.parse((*specs, (), None))
+        return tree_unflatten(out_plan.user_outputs(op(args)), structure)
+
+    return call
+
+
+def register_custom_op_with_autograd(
     *,
     op_name: str,
     input_tensors_for_grad: List[str],
@@ -1226,12 +1352,12 @@ def register_custom_op(
     ``bwd_arg_type``.
 
     Registration touches experimental ``torch.library`` / opaque-object APIs
-    that may be missing on older PyTorch. If it fails, this warns once and
-    returns ``None`` instead of raising, so callers can fall back to eager under
-    ``torch.compile`` (a graph break) rather than breaking import.
+    that may be missing on older PyTorch. Unavailable or incompatible APIs cause
+    a warning and return ``None``, allowing eager execution under ``torch.compile``
+    (a graph break). Duplicate operator names raise ``ValueError``.
     """
     try:
-        return _register_custom_op_impl(
+        return _register_custom_op_with_autograd_impl(
             op_name=op_name,
             input_tensors_for_grad=input_tensors_for_grad,
             fwd_arg_type=fwd_arg_type,
@@ -1249,7 +1375,7 @@ def register_custom_op(
         return None
 
 
-def _register_custom_op_impl(
+def _register_custom_op_with_autograd_impl(
     *,
     op_name: str,
     input_tensors_for_grad: List[str],
@@ -1261,7 +1387,7 @@ def _register_custom_op_impl(
     fwd_fake_impl: Callable[[Any], Tuple[Any, ...]],
     bwd_fake_impl: Callable[[Any], Tuple[Any, ...]],
 ) -> Callable[..., Any]:
-    """Body of :func:`register_custom_op`; see it for semantics."""
+    """Body of :func:`register_custom_op_with_autograd`; see it for semantics."""
     # Existence check at the API boundary: every ``input_tensors_for_grad`` name
     # must be an actual field of ``fwd_arg_type`` (differentiability -- whether
     # that field can carry a gradient -- is checked later, in
@@ -1271,96 +1397,39 @@ def _register_custom_op_impl(
     if missing:
         raise ValueError(f"input_tensors_for_grad names not in {fwd_arg_type.__name__}: {missing}")
 
-    wrapper_fwd_name = op_name
-    wrapper_bwd_name = f"{op_name}_backward"
-    base_fwd_name = f"{op_name}_base"
-    base_bwd_name = f"{wrapper_bwd_name}_base"
-    subclass_list = _all_quantized_tensor_subclasses()
-
-    fwd_plan = _parse_arg_type(fwd_arg_type)
-    bwd_plan = _parse_arg_type(bwd_arg_type)
-
-    num_grad_inputs = len(input_tensors_for_grad)
-    grad_targets = fwd_plan.resolve_grad_targets(input_tensors_for_grad)
-
-    fwd_schema = f"{fwd_plan.schema_str} -> Tensor[]"
-    bwd_schema = f"{bwd_plan.schema_str} -> Tensor[]"
-
-    base_bwd_qualname = f"{_TE_OP_NAMESPACE}::{base_bwd_name}"
-
-    base_fwd_def = _register_base_op(
-        op_name=base_fwd_name,
-        schema_str=fwd_schema,
-        plan=fwd_plan,
+    fwd_op = _register_op(
+        name=op_name,
+        arg_type=fwd_arg_type,
         impl=fwd_impl,
         fake_impl=fwd_fake_impl,
         pack_result=_pack_fwd_result,
     )
-    _register_base_op(
-        op_name=base_bwd_name,
-        schema_str=bwd_schema,
-        plan=bwd_plan,
+    bwd_qualname = f"{_TE_OP_NAMESPACE}::{op_name}_backward_base"
+    num_grad_inputs = len(input_tensors_for_grad)
+    bwd_op = _register_op(
+        name=f"{op_name}_backward",
+        arg_type=bwd_arg_type,
         impl=bwd_impl,
         fake_impl=bwd_fake_impl,
-        pack_result=lambda g: _pack_bwd_result(g, num_grad_inputs, base_bwd_qualname),
-    )
-
-    base_fwd_op = getattr(getattr(torch.ops, _TE_OP_NAMESPACE), base_fwd_name)
-    base_bwd_op = getattr(getattr(torch.ops, _TE_OP_NAMESPACE), base_bwd_name)
-
-    fwd_slot_offsets = fwd_plan.tensor_or_quantized_offsets()
-    bwd_slot_offsets = bwd_plan.tensor_or_quantized_offsets()
-
-    wrapper_fwd_def = _register_wrapper_op(
-        wrapper_op_name=wrapper_fwd_name,
-        schema_str=fwd_schema,
-        base_op=base_fwd_op,
-        slot_offsets=fwd_slot_offsets,
-        subclasses=subclass_list,
-    )
-    # Pass-through: a subclass input reaches the base op through the dispatch
-    # rule below, never through the wrapper body.
-    wrapper_bwd_def = _register_wrapper_op(
-        wrapper_op_name=wrapper_bwd_name, schema_str=bwd_schema, base_op=base_bwd_op
+        pack_result=lambda grads: _pack_bwd_result(grads, num_grad_inputs, bwd_qualname),
     )
 
     autograd_common = {
-        "fwd_plan": fwd_plan,
-        "bwd_plan": bwd_plan,
-        "grad_targets": grad_targets,
+        "fwd_plan": fwd_op.plan,
+        "bwd_plan": bwd_op.plan,
+        "grad_targets": fwd_op.plan.resolve_grad_targets(input_tensors_for_grad),
         "setup_context_user": setup_context,
         "fwd_fake_impl": fwd_fake_impl,
     }
-    wrapper_fwd_op = getattr(getattr(torch.ops, _TE_OP_NAMESPACE), wrapper_fwd_name)
-    wrapper_bwd_op = getattr(getattr(torch.ops, _TE_OP_NAMESPACE), wrapper_bwd_name)
-
-    _register_autograd_for_op(fwd_op=base_fwd_def, bwd_op=base_bwd_op, **autograd_common)
-    _register_autograd_for_op(fwd_op=wrapper_fwd_def, bwd_op=wrapper_bwd_op, **autograd_common)
-
-    _fwd_rule = _make_dispatch_rule(
-        _make_slot_forwarder(base_fwd_op, fwd_slot_offsets, subclass_list)
+    _register_autograd_for_op(fwd_op=fwd_op.base_def, bwd_op=bwd_op.base_op, **autograd_common)
+    _register_autograd_for_op(
+        fwd_op=fwd_op.wrapper_def, bwd_op=bwd_op.wrapper_op, **autograd_common
     )
-    _bwd_rule = _make_dispatch_rule(
-        _make_slot_forwarder(base_bwd_op, bwd_slot_offsets, subclass_list)
-    )
-
-    for sub in subclass_list:
-        wrapper_fwd_def.register_torch_dispatch(sub, _fwd_rule)
-        wrapper_bwd_def.register_torch_dispatch(sub, _bwd_rule)
-
-    _quantized_tensor_passthrough_ops.add(wrapper_fwd_op.default)
-    _quantized_tensor_passthrough_ops.add(wrapper_bwd_op.default)
-    _quantized_tensor_passthrough_ops.add(base_fwd_op.default)
-    _quantized_tensor_passthrough_ops.add(base_bwd_op.default)
 
     def forward_fn(fwd_args):
-        spec_obj = _spec_view(fwd_args, fwd_plan.tensor_field_names())
-        out_plan = _OutputPlan.parse(fwd_fake_impl(spec_obj))
-        kwargs = fwd_plan.pack(fwd_args)
-        flat_in = [kwargs[name] for name in fwd_plan.slot_names]
-        result = wrapper_fwd_op(*flat_in)
-
-        outputs = out_plan.user_outputs(result)
+        spec_args = _spec_view(fwd_args, fwd_op.plan.tensor_field_names())
+        out_plan = _OutputPlan.parse(fwd_fake_impl(spec_args))
+        outputs = out_plan.user_outputs(fwd_op(fwd_args))
         if len(outputs) == 1:
             return outputs[0]
         return tuple(outputs)
diff --git a/transformer_engine/pytorch/module/linear.py b/transformer_engine/pytorch/module/linear.py
index 94de69e975..b6a25cc37f 100644
--- a/transformer_engine/pytorch/module/linear.py
+++ b/transformer_engine/pytorch/module/linear.py
@@ -88,7 +88,7 @@
 from ..dynamo import (
     TensorSpec,
     TensorOrQuantized,
-    register_custom_op,
+    register_custom_op_with_autograd,
     is_value_opaque_quantizer,
 )
 from ..tensor.float8_tensor import Float8CurrentScalingQuantizer, Float8Quantizer
@@ -1788,7 +1788,7 @@ def _linear_backward_fake(
 
 
 # Custom op used under ``torch.compile``.
-_linear_op = register_custom_op(
+_linear_op = register_custom_op_with_autograd(
     op_name="linear",
     input_tensors_for_grad=["weight", "inp", "bias"],
     fwd_arg_type=LinearFwdArgs,
diff --git a/transformer_engine/pytorch/ops/fuser.py b/transformer_engine/pytorch/ops/fuser.py
index 2500002700..e5e40e2945 100644
--- a/transformer_engine/pytorch/ops/fuser.py
+++ b/transformer_engine/pytorch/ops/fuser.py
@@ -6,13 +6,15 @@
 
 from __future__ import annotations
 from collections.abc import Callable, Iterable, Sequence
+import copy
 import itertools
 from typing import Any, Optional, TypeAlias
 
 import torch
 
-from ..quantization import FP8GlobalStateManager, Recipe
+from ..quantization import FP8GlobalStateManager, Recipe, _has_delayed_scaling_state
 from ..quantized_tensor import prepare_for_saving, restore_from_func_ctx
+from ..utils import warn_compile_eager_fallback
 from .op import (
     BasicOperation,
     FusibleOperation,
@@ -68,6 +70,7 @@ def forward(
         fuser: OperationFuser,
         basic_op_kwargs: list[dict[str, Any]],
         set_output_requires_grad: bool,
+        use_custom_ops: bool,
         *params_and_extra_inputs: torch.Tensor,
     ) -> torch.Tensor | tuple[torch.Tensor, ...]:
         """Forward pass
@@ -84,6 +87,9 @@ def forward(
             Keyword arguments to BasicOperation
         set_output_requires_grad: bool
             Whether to set ``requires_grad`` flags on returned tensors
+        use_custom_ops: bool
+            Whether to call the operations' custom ops instead of tracing their
+            eager implementations. Decided once per group by ``OperationFuser``.
         *params_and_extra_inputs: torch.Tensor
             Other tensor inputs to include in autograd graph. Consists
             of parameter tensors, followed by extra operation inputs.
@@ -100,9 +106,14 @@ def forward(
         # Operation autograd contexts
         basic_op_ctxs = [OperationContext() for _ in range(fuser._num_basic_ops)]
 
-        # Mark input tensors as not deletable in backward
-        for tensor in (input_,) + params_and_extra_inputs:
-            tensor._do_not_clear = True
+        # Mark input tensors as not deletable in backward. Skipped whenever this
+        # is being traced -- not merely when the custom ops are used: these
+        # tensors are created outside this function, and a higher-order op may
+        # not mutate anything from an enclosing scope. Under fullgraph there is
+        # no falling back out of the graph, so the constraint holds either way.
+        if not torch.compiler.is_compiling():
+            for tensor in (input_,) + params_and_extra_inputs:
+                tensor._do_not_clear = True
 
         # Place user provided extra inputs into their basic-op slots. Slots bound to
         # internal channels are filled lazily as their producers execute.
@@ -155,6 +166,7 @@ def forward(
             if next_op is not None:
                 next_op_input_quantizer = next_op.get_input_quantizer()
 
+            compile_kwargs = {"use_custom_ops": True} if use_custom_ops else {}
             x, fused_op_extra_outputs = op.fuser_forward(
                 [basic_op_ctxs[idx] for idx in basic_op_idxs],
                 x,
@@ -162,6 +174,7 @@ def forward(
                 prev_op_grad_output_quantizer=prev_op_grad_output_quantizer,
                 next_op_input_quantizer=next_op_input_quantizer,
                 basic_op_kwargs=[basic_op_kwargs[idx] for idx in basic_op_idxs],
+                **compile_kwargs,
             )
             if len(fused_op_extra_outputs) != len(basic_op_idxs):
                 raise RuntimeError(
@@ -229,9 +242,13 @@ def forward(
             func_ctx.save_for_backward(*tensors_to_save)
             func_ctx.tensor_objects = tensor_objects
 
-            # Whether to perform recipe update in backward pass
+            # Whether to perform recipe update in backward pass. Skipped under
+            # compile: this reads and flips global FP8 state, and delayed
+            # scaling -- the only recipe it serves -- is gated out anyway.
             is_first_module = False
-            if fuser.first_op_requiring_backward < fuser._num_basic_ops:
+            if not torch.compiler.is_compiling() and (
+                fuser.first_op_requiring_backward < fuser._num_basic_ops
+            ):
                 is_first_module = FP8GlobalStateManager.is_first_fp8_module()
 
             # Other context
@@ -246,15 +263,20 @@ def forward(
             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
-
-        # Mark output tensors as not deletable in backward
-        for tensor in itertools.chain(
-            (x,),
-            (y for ys in extra_outputs for y in ys if y is not None),
-        ):
-            tensor._do_not_clear = True
-
-        if set_output_requires_grad:
+            func_ctx.use_custom_ops = use_custom_ops
+
+        # Mark output tensors as not deletable in backward (eager only; see above)
+        if not torch.compiler.is_compiling():
+            for tensor in itertools.chain(
+                (x,),
+                (y for ys in extra_outputs for y in ys if y is not None),
+            ):
+                tensor._do_not_clear = True
+
+        # Autograd marks the outputs of an ``apply`` itself, so this is only
+        # needed on the eager path -- and AOTAutograd's functionalization drops
+        # a requires_grad_() applied to a graph output anyway.
+        if set_output_requires_grad and not torch.compiler.is_compiling():
             x.requires_grad_(fuser.first_op_requiring_backward < fuser._num_basic_ops)
 
         if extra_outputs_flat:
@@ -279,7 +301,12 @@ def backward(
         # Restore saved tensors
         saved_tensors = restore_from_func_ctx(func_ctx)
 
-        # Unflatten list of saved tensors
+        # Unflatten list of saved tensors. Under compile the contexts were
+        # created in the forward, which is a different subgraph, so writing to
+        # them here would be a side effect on an enclosing scope; copy them into
+        # this one instead. The copy carries the attributes the forward set.
+        if torch.compiler.is_compiling():
+            basic_op_ctxs = [copy.copy(ctx) for ctx in basic_op_ctxs]
         for ctx in basic_op_ctxs:
             ctx.saved_tensors = saved_tensors[slice(*ctx._saved_tensors_range)]
             ctx._saved_tensors_range = None
@@ -329,10 +356,12 @@ def backward(
                                 channel_grad if output_grad is None else output_grad + channel_grad
                             )
             grad_extra_outputs = [basic_op_grad_extra_outputs[idx] for idx in basic_op_idxs]
+            compile_kwargs = {"use_custom_ops": True} if func_ctx.use_custom_ops else {}
             dx, fused_op_grad_params, fused_op_grad_extra_inputs = op.fuser_backward(
                 [basic_op_ctxs[idx] for idx in basic_op_idxs],
                 dx,
                 basic_op_grad_extra_outputs=grad_extra_outputs,
+                **compile_kwargs,
             )
             for idx, dparams in zip(basic_op_idxs, fused_op_grad_params):
                 grad_params[idx] = dparams
@@ -394,6 +423,7 @@ def backward(
             None,  # fuser
             None,  # basic_op_kwargs
             None,  # set_output_requires_grad
+            None,  # use_custom_ops
             *grad_params_flat,
             *grad_extra_inputs_flat,
         )
@@ -707,6 +737,49 @@ def maybe_fuse_ops(
         # state, so the mapped lists can be selected directly on cache hits.
         self._fused_ops_cache[fusion_params] = (self._forward_ops, self._backward_ops)
 
+    def _custom_ops_unsupported_reason(
+        self, basic_op_kwargs: list[dict[str, Any]]
+    ) -> Optional[str]:
+        """Why this group may not run through its operations' custom ops."""
+        for mode, ops in (("forward", self._forward_ops), ("backward", self._backward_ops)):
+            if len(ops) != self._num_basic_ops or any(
+                op is not self._basic_ops[idx] or basic_op_idxs != [idx]
+                for idx, (op, basic_op_idxs) in enumerate(ops)
+            ):
+                return f"a {mode} fusion"
+        if self._num_basic_ops != 1:
+            return "a group of several operations"
+        for op, kwargs in zip(self._basic_ops, basic_op_kwargs, strict=True):
+            # Only tensors. The other fields of an args container are values read
+            # off the module, constant across calls and baked into the graph; a
+            # kwarg changes per call, and on the second value Dynamo hands over a
+            # symbolic scalar, which cannot go into an opaque value bundle. Pass a
+            # 0-d tensor instead -- it is a graph input, so it does not recompile.
+            values = sorted(name for name, v in kwargs.items() if not isinstance(v, torch.Tensor))
+            if values:
+                return f"{type(op).__name__} with non-tensor keyword arguments {values}"
+        for op in self._basic_ops:
+            if op.num_extra_inputs or op.num_extra_outputs:
+                return f"{type(op).__name__} with extra tensor inputs or outputs"
+            reason = op.compile_unsupported_reason()
+            if reason is not None:
+                return reason
+        return None
+
+    def _use_custom_ops(self, basic_op_kwargs: list[dict[str, Any]]) -> bool:
+        """Whether this group runs through its operations' custom ops.
+
+        Decided once for the whole group: a pipeline compiles as a whole, so one
+        unsupported operation sends all of them to eager.
+        """
+        if not torch.compiler.is_compiling():
+            return False
+        reason = self._custom_ops_unsupported_reason(basic_op_kwargs)
+        if reason is None:
+            return True
+        warn_compile_eager_fallback(reason)
+        return False
+
     def __call__(
         self,
         input: torch.Tensor,  # pylint: disable=redefined-builtin
@@ -743,17 +816,29 @@ def __call__(
 
         # Initialization before forward
         for idx, op in enumerate(self._basic_ops):
+            if torch.compiler.is_compiling() and op._fp8_metas is not None:
+                if any(
+                    meta is not None and _has_delayed_scaling_state(meta)
+                    for meta in op._fp8_metas.values()
+                ):
+                    raise RuntimeError(
+                        "Delayed scaling is not supported under torch.compile in OperationFuser, "
+                        "including CustomRecipe with DelayedScalingRequest."
+                    )
             op.pre_fuser_forward(requires_grad=idx >= self.first_op_requiring_backward)
 
         # Fuser forward pass
         # Note: We call forward directly when is_grad_enabled=False,
         # which can expose non-leaf tensors to the inner ops. Avoid
         # problems in this case by passing set_output_requires_grad=False.
+        use_custom_ops = self._use_custom_ops(basic_op_kwargs)
+
         args = (
             input,
             self,
             basic_op_kwargs,
             is_grad_enabled,  # set_output_requires_grad
+            use_custom_ops,
             *self._flat_basic_op_params,
             *extra_inputs,
         )
diff --git a/transformer_engine/pytorch/ops/op.py b/transformer_engine/pytorch/ops/op.py
index d057d46816..5968d66430 100644
--- a/transformer_engine/pytorch/ops/op.py
+++ b/transformer_engine/pytorch/ops/op.py
@@ -9,7 +9,7 @@
 from collections.abc import Iterable, Sequence
 import dataclasses
 import pickle
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 import torch
 
@@ -22,6 +22,7 @@
     autocast,
 )
 from ..tensor import Quantizer
+from ..dynamo import is_value_opaque_quantizer, register_custom_op
 
 
 @dataclasses.dataclass
@@ -59,6 +60,55 @@ def save_for_backward(self, *tensors: Optional[torch.Tensor]) -> None:
 class FusibleOperation(torch.nn.Module, metaclass=abc.ABCMeta):
     """Tensor operation supported by the operation fuser"""
 
+    # Custom ops are registered once per operation class.
+    fwd_args_type: Optional[type] = None
+    bwd_args_type: Optional[type] = None
+    # (forward_fn, backward_fn), or None if the operation cannot be compiled.
+    compile_ops: Optional[tuple[Callable[..., Any], Callable[..., Any]]] = None
+
+    def __init_subclass__(cls, **kwargs) -> None:
+        super().__init_subclass__(**kwargs)
+        if cls.fwd_args_type is not None and cls.bwd_args_type is not None:
+            cls._register_compile_ops()
+
+    @classmethod
+    def _register_compile_ops(cls) -> None:
+        for name in ("fwd_args_type", "bwd_args_type"):
+            if not dataclasses.is_dataclass(getattr(cls, name)):
+                raise TypeError(f"{cls.__name__}.{name} must be a dataclass")
+        name = cls.__name__.lower()
+        forward = register_custom_op(
+            op_name=name,
+            arg_type=cls.fwd_args_type,
+            impl=cls.forward_compute,
+            fake_impl=cls.forward_fake,
+        )
+        backward = register_custom_op(
+            op_name=f"{name}_backward",
+            arg_type=cls.bwd_args_type,
+            impl=cls.backward_compute,
+            fake_impl=cls.backward_fake,
+        )
+        cls.compile_ops = (
+            (forward, backward) if forward is not None and backward is not None else None
+        )
+
+    def compile_unsupported_reason(self) -> Optional[str]:
+        """Why this operation cannot run through its custom op, or ``None``."""
+        if self.compile_ops is None:
+            return f"{self.__class__.__name__} without a custom op"
+        basic_ops = self.basic_ops if self.is_fused_op else (self,)
+        for op in basic_ops:
+            for mode in ("forward", "backward"):
+                for index in range(op.num_quantizers(mode)):
+                    quantizer = op.get_quantizer(mode, index)
+                    if quantizer is not None and not is_value_opaque_quantizer(quantizer):
+                        return (
+                            f"{type(quantizer).__name__} (not a torch.compile value-opaque"
+                            " quantizer)"
+                        )
+        return None
+
     @property
     @abc.abstractmethod
     def is_fused_op(self) -> bool:
@@ -89,6 +139,7 @@ def fuser_forward(
         prev_op_grad_output_quantizer: Optional[Quantizer],
         next_op_input_quantizer: Optional[Quantizer],
         basic_op_kwargs: list[dict[str, Any]],
+        use_custom_ops: bool = False,
     ) -> tuple[torch.Tensor, Sequence[Sequence[Optional[torch.Tensor]]]]:
         """Forward pass
 
@@ -124,9 +175,19 @@ def fuser_forward(
             channel owned by this fused operation may be ``None``.
 
         """
-        raise NotImplementedError(
-            f"Forward pass is not implemented for operation ({self.__class__.__name__})"
+        args = self.pack_forward_args(
+            basic_op_ctxs,
+            input_,
+            basic_op_extra_inputs=basic_op_extra_inputs,
+            prev_op_grad_output_quantizer=prev_op_grad_output_quantizer,
+            next_op_input_quantizer=next_op_input_quantizer,
+            basic_op_kwargs=basic_op_kwargs,
         )
+        compute = self.compile_ops[0] if use_custom_ops else self.forward_compute
+        output, extra_outputs, aux = compute(args)
+        if any(ctx.requires_grad for ctx in basic_op_ctxs):
+            self.forward_setup_context(basic_op_ctxs, args, aux)
+        return output, extra_outputs
 
     def fuser_backward(
         self,
@@ -134,6 +195,7 @@ def fuser_backward(
         grad_output: torch.Tensor,
         *,
         basic_op_grad_extra_outputs: Sequence[Sequence[Optional[torch.Tensor]]],
+        use_custom_ops: bool = False,
     ) -> tuple[
         torch.Tensor,
         Sequence[Sequence[Optional[torch.Tensor]]],
@@ -168,9 +230,67 @@ def fuser_backward(
             operations
 
         """
-        raise NotImplementedError(
-            f"Backward pass is not implemented for operation ({self.__class__.__name__})"
+        args = self.pack_backward_args(
+            basic_op_ctxs,
+            grad_output,
+            basic_op_grad_extra_outputs=basic_op_grad_extra_outputs,
         )
+        compute = self.compile_ops[1] if use_custom_ops else self.backward_compute
+        grad_input, grad_params, grad_extra_inputs = compute(args)
+        if grad_input is None:
+            grad_input = grad_output
+        return grad_input, grad_params, grad_extra_inputs
+
+    @classmethod
+    def forward_compute(cls, args: Any) -> tuple:
+        """Return (output, extra_outputs per basic op, fresh aux tensors)."""
+        raise NotImplementedError
+
+    @classmethod
+    def forward_fake(cls, args: Any) -> tuple:
+        """Shape-only twin of forward_compute, using TensorSpec."""
+        raise NotImplementedError
+
+    @classmethod
+    def backward_compute(cls, args: Any) -> tuple:
+        """Return (grad_input, grad_params per basic op, grad_extra_inputs per basic op).
+
+        A None grad_input passes grad_output through unchanged.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def backward_fake(cls, args: Any) -> tuple:
+        """Shape-only twin of backward_compute, using TensorSpec."""
+        raise NotImplementedError
+
+    def pack_forward_args(
+        self,
+        basic_op_ctxs: list[OperationContext],
+        input_: torch.Tensor,
+        *,
+        basic_op_extra_inputs: Sequence[Sequence[Optional[torch.Tensor]]],
+        prev_op_grad_output_quantizer: Optional[Quantizer],
+        next_op_input_quantizer: Optional[Quantizer],
+        basic_op_kwargs: list[dict[str, Any]],
+    ) -> Any:
+        """Gather inputs and module state into fwd_args_type."""
+        raise NotImplementedError
+
+    def pack_backward_args(
+        self,
+        basic_op_ctxs: list[OperationContext],
+        grad_output: torch.Tensor,
+        *,
+        basic_op_grad_extra_outputs: Sequence[Sequence[Optional[torch.Tensor]]],
+    ) -> Any:
+        """Gather gradients and saved context into bwd_args_type."""
+        raise NotImplementedError
+
+    def forward_setup_context(
+        self, basic_op_ctxs: list[OperationContext], args: Any, aux: tuple
+    ) -> None:
+        """Save state needed by the basic operations' backward passes."""
 
 
 class BasicOperation(FusibleOperation, metaclass=abc.ABCMeta):
@@ -508,7 +628,6 @@ def _load_fp8_metas(self, fp8_metas: Optional[dict[str, Any]]) -> None:
                 self._fp8_metas[mode][fp8_meta_key].scale.copy_(scale)
                 self._fp8_metas[mode][fp8_meta_key].amax_history.copy_(amax_history)
 
-    @abc.abstractmethod
     def op_forward(
         self,
         ctx: OperationContext,
@@ -520,6 +639,8 @@ def op_forward(
     ) -> torch.Tensor:
         """Forward pass
 
+        Convenience interface for operations without extra tensor inputs or outputs.
+
         Parameters
         ----------
         ctx: OperationContext
@@ -537,8 +658,8 @@ def op_forward(
             Output tensor
 
         """
+        raise NotImplementedError
 
-    @abc.abstractmethod
     def op_backward(
         self,
         ctx: OperationContext,
@@ -546,6 +667,8 @@ def op_backward(
     ) -> tuple[torch.Tensor, Iterable[Optional[torch.Tensor]]]:
         """Backward pass
 
+        Convenience interface for operations without extra tensor inputs or outputs.
+
         Parameters
         ----------
         ctx: OperationContext
@@ -561,6 +684,7 @@ def op_backward(
             Loss gradients w.r.t. parameters
 
         """
+        raise NotImplementedError
 
     def fuser_forward(
         self,
@@ -571,7 +695,18 @@ def fuser_forward(
         prev_op_grad_output_quantizer: Optional[Quantizer],
         next_op_input_quantizer: Optional[Quantizer],
         basic_op_kwargs: list[dict[str, Any]],
-    ) -> tuple[torch.Tensor, list[tuple[()]]]:
+        use_custom_ops: bool = False,
+    ) -> tuple[torch.Tensor, Sequence[Sequence[Optional[torch.Tensor]]]]:
+        if use_custom_ops or type(self).op_forward is BasicOperation.op_forward:
+            return super().fuser_forward(
+                basic_op_ctxs,
+                input_,
+                basic_op_extra_inputs=basic_op_extra_inputs,
+                prev_op_grad_output_quantizer=prev_op_grad_output_quantizer,
+                next_op_input_quantizer=next_op_input_quantizer,
+                basic_op_kwargs=basic_op_kwargs,
+                use_custom_ops=use_custom_ops,
+            )
         if self.num_extra_inputs > 0 or self.num_extra_outputs > 0:
             raise RuntimeError(
                 "{self.__class__.__name__} operation has "
@@ -594,11 +729,19 @@ def fuser_backward(
         grad_output: torch.Tensor,
         *,
         basic_op_grad_extra_outputs: list[tuple[torch.Tensor, ...]],
+        use_custom_ops: bool = False,
     ) -> tuple[
         torch.Tensor,
-        list[Iterable[Optional[torch.Tensor]]],
-        list[tuple[()]],
+        Sequence[Sequence[Optional[torch.Tensor]]],
+        Sequence[Sequence[Optional[torch.Tensor]]],
     ]:
+        if use_custom_ops or type(self).op_backward is BasicOperation.op_backward:
+            return super().fuser_backward(
+                basic_op_ctxs,
+                grad_output,
+                basic_op_grad_extra_outputs=basic_op_grad_extra_outputs,
+                use_custom_ops=use_custom_ops,
+            )
         if self.num_extra_inputs > 0 or self.num_extra_outputs > 0:
             raise RuntimeError(
                 "{self.__class__.__name__} operation has "
diff --git a/transformer_engine/pytorch/quantized_tensor.py b/transformer_engine/pytorch/quantized_tensor.py
index 7149a5a163..4d64bc43c7 100644
--- a/transformer_engine/pytorch/quantized_tensor.py
+++ b/transformer_engine/pytorch/quantized_tensor.py
@@ -282,11 +282,16 @@ def __tensor_unflatten__(
         )
 
 
+class _SavedQuantizedTensor(NamedTuple):
+    inner_names: tuple[str, ...]
+    metadata: Dict[str, Any]
+
+
 def prepare_for_saving(
     *tensors: Union[torch.Tensor, QuantizedTensorStorage],
 ) -> Tuple[
     list[Optional[Union[torch.Tensor, torch.nn.Parameter]]],
-    list[Optional[QuantizedTensorStorage]],
+    list[Optional[Union[QuantizedTensorStorage, _SavedQuantizedTensor]]],
 ]:
     """Prepare tensors for saving. Needed because save_for_backward accepts only
     torch.Tensor/torch.nn.Parameter types, while we want to be able to save
@@ -297,6 +302,10 @@ def prepare_for_saving(
         if tensor is None or isinstance(tensor, torch.Tensor):
             tensor_list.append(tensor)
             tensor_objects_list.append(None)
+        elif torch.compiler.is_compiling():
+            inner_names, metadata = tensor.__tensor_flatten__()
+            tensor_list.extend(getattr(tensor, name) for name in inner_names)
+            tensor_objects_list.append(_SavedQuantizedTensor(tuple(inner_names), metadata))
         else:
             t, t_obj = tensor.prepare_for_saving()
             tensor_list.extend(t)
@@ -306,7 +315,7 @@ def prepare_for_saving(
 
 
 def restore_from_saved(
-    tensors: list[Optional[Union[torch.Tensor, QuantizedTensorStorage]]],
+    tensors: list[Optional[Union[torch.Tensor, QuantizedTensorStorage, _SavedQuantizedTensor]]],
     saved_tensors: list[Optional[Union[torch.Tensor, torch.nn.Parameter]]],
     return_saved_tensors: bool = False,
 ) -> (
@@ -324,6 +333,13 @@ def restore_from_saved(
         if tensor is None or isinstance(tensor, torch.Tensor):
             tensor_objects.append(saved_tensors[0])
             saved_tensors = saved_tensors[1:]
+        elif isinstance(tensor, _SavedQuantizedTensor):
+            count = len(tensor.inner_names)
+            inner = dict(zip(tensor.inner_names, saved_tensors[:count]))
+            tensor_objects.append(
+                QuantizedTensorStorage.__tensor_unflatten__(inner, tensor.metadata, None, None)
+            )
+            saved_tensors = saved_tensors[count:]
         else:
             saved_tensors = tensor.restore_from_saved(saved_tensors)
             tensor_objects.append(tensor)
diff --git a/transformer_engine/pytorch/utils.py b/transformer_engine/pytorch/utils.py
index c4c6024f47..00bc627031 100644
--- a/transformer_engine/pytorch/utils.py
+++ b/transformer_engine/pytorch/utils.py
@@ -131,6 +131,8 @@ def clear_tensor_data(*tensors: Tuple[Optional[torch.Tensor], ...]) -> None:
 
     Must be used carefully.
     """
+    if torch.compiler.is_compiling():
+        return
 
     for t in tensors:
         if t is not None: