diff --git a/.github/benchmark/models_accuracy.json b/.github/benchmark/models_accuracy.json
index 554441baf4..4eb18e3933 100644
--- a/.github/benchmark/models_accuracy.json
+++ b/.github/benchmark/models_accuracy.json
@@ -249,7 +249,7 @@
   {
     "model_name": "Kimi-K3",
     "model_path": "moonshotai/Kimi-K3",
-    "extraArgs": "--kv_cache_dtype fp8 -tp 8 --trust-remote-code --gpu-memory-utilization 0.93 --block-size 128 --no-enable_prefix_caching",
+    "extraArgs": "--kv_cache_dtype fp8 -tp 8 --trust-remote-code --gpu-memory-utilization 0.93 --block-size 128 --no-enable_prefix_caching --online_quant_config '{\"global_quant_config\": \"ptpc_fp8\", \"exclude_layer\": [\"lm_head\", \"model.embed_tokens\", \"*self_attn.[qkv]_conv1d*\", \"*block_sparse_moe.experts*\", \"*block_sparse_moe.routed_expert_*\", \"*vision_tower*\", \"*mm_projector*\"]}'",
     "env_vars": "",
     "runner": "linux-atom-do-mi350x-8",
     "test_level": "pr",
diff --git a/atom/config.py b/atom/config.py
index 021063dde9..a6e78f691c 100644
--- a/atom/config.py
+++ b/atom/config.py
@@ -314,8 +314,15 @@ def __init__(
             "mxfp4",
             "mxfp8",
             "quark",
+            "compressed-tensors",
         ]:
             self.online_quant = True
+            if self.quant_method == "compressed-tensors":
+                logger.warning(
+                    "Online quant with compressed-tensors is not fully supported. "
+                    "Be careful about the online quant config setting when launching "
+                    "the server."
+                )
             online_parser = get_quant_parser("online_quant")
             online_parsed_quant_config = online_parser.parse(online_quant_config)
             self.online_global_spec = online_parsed_quant_config.global_spec
diff --git a/atom/model_ops/kimi_k3/__init__.py b/atom/model_ops/kimi_k3/__init__.py
index 92c1bf21ba..b1aeccc93d 100644
--- a/atom/model_ops/kimi_k3/__init__.py
+++ b/atom/model_ops/kimi_k3/__init__.py
@@ -3,7 +3,10 @@
 
 """Fused model operations for Kimi-K3."""
 
-from atom.model_ops.kimi_k3.activations import rmsnorm_gated, situ_and_mul
+from atom.model_ops.kimi_k3.activations import (
+    rmsnorm_gated,
+    situ_and_mul,
+)
 from atom.model_ops.kimi_k3.attention_residual import apply_attn_res
 from atom.model_ops.kimi_k3.kda_state import gather_kda_initial_state
 
diff --git a/atom/model_ops/kimi_k3/activations.py b/atom/model_ops/kimi_k3/activations.py
index b4a520e902..9708abd039 100644
--- a/atom/model_ops/kimi_k3/activations.py
+++ b/atom/model_ops/kimi_k3/activations.py
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import torch
+from aiter import QuantType, dtypes, get_hip_quant
 
 try:
     import triton
@@ -81,6 +82,53 @@ def _rmsnorm_gated_kernel(
             y_ptr + row * stride_ym + cols, y.to(y_ptr.dtype.element_ty), mask=mask
         )
 
+    @triton.jit
+    def _rmsnorm_gated_fp8_per_token_kernel(
+        x_ptr,
+        w_ptr,
+        g_ptr,
+        y_ptr,
+        s_ptr,
+        H,
+        eps,
+        fp8_max,
+        stride_xm,
+        stride_xh,
+        stride_g_outer,
+        stride_g_head,
+        stride_ym,
+        HEADS: tl.constexpr,
+        HEADS_POW2: tl.constexpr,
+        BLOCK: tl.constexpr,
+    ):
+        tok = tl.program_id(0)
+        head_ids = tl.arange(0, HEADS_POW2)
+        cols = tl.arange(0, BLOCK)
+        mask = (head_ids[:, None] < HEADS) & (cols[None, :] < H)  # [HEADS_POW2, BLOCK]
+        # Padding heads (head_ids >= HEADS) are masked out on every load/store,
+        # but their raw offset (head_ids * stride) can still address past the end
+        # of the buffer -- forming an out-of-bounds pointer is UB on ROCm/triton
+        # and faults when the allocation abuts an unmapped page. Clamp the head
+        # index used for addressing to a valid row; the mask (other=0.0) still
+        # discards the value, so numerics are unchanged.
+        h_safe = tl.where(head_ids < HEADS, head_ids, 0)
+        x_off = tok * stride_xm + h_safe[:, None] * stride_xh + cols[None, :]
+        x = tl.load(x_ptr + x_off, mask=mask, other=0.0).to(tl.float32)
+        var = tl.sum(x * x, axis=1) / H  # [HEADS]
+        rstd = 1.0 / tl.sqrt(var + eps)  # [HEADS]
+        w = tl.load(w_ptr + cols, mask=cols < H, other=0.0).to(tl.float32)  # [BLOCK]
+        g_off = tok * stride_g_outer + h_safe[:, None] * stride_g_head + cols[None, :]
+        gate = tl.load(g_ptr + g_off, mask=mask, other=0.0).to(tl.float32)
+        normed = (x * rstd[:, None] * w[None, :]) * tl.sigmoid(gate)  # [HEADS, BLOCK]
+        amax = tl.max(tl.abs(normed))  # scalar per token
+        scale = amax / fp8_max
+        inv = tl.where(scale > 0.0, 1.0 / scale, 0.0)
+        q = normed * inv
+        q = tl.minimum(tl.maximum(q, -fp8_max), fp8_max)
+        y_off = tok * stride_ym + h_safe[:, None] * H + cols[None, :]
+        tl.store(y_ptr + y_off, q.to(y_ptr.dtype.element_ty), mask=mask)
+        tl.store(s_ptr + tok, scale)
+
 
 def situ_and_mul(
     x: torch.Tensor, beta: float, linear_beta: float | None
@@ -115,15 +163,36 @@ def situ_and_mul(
 
 
 def rmsnorm_gated(
-    x: torch.Tensor, weight: torch.Tensor, gate: torch.Tensor, eps: float
-) -> torch.Tensor:
+    x: torch.Tensor,
+    weight: torch.Tensor,
+    gate: torch.Tensor,
+    eps: float,
+    quant_type: QuantType | None = None,
+    quant_dtype: torch.dtype | None = None,
+) -> torch.Tensor | tuple[torch.Tensor, torch.Tensor]:
     """rmsnorm(x) over last dim * weight * sigmoid(gate).
 
+    When ``(quant_type, quant_dtype)`` is the per-token FP8 scheme, the normed
+    output is also quantized and the function returns
+    ``(fp8 [t, heads*H], scale [t, 1])`` ready for the consuming GEMM's
+    ``x_scale=`` path. With no quant (``None``/``QuantType.No``) it returns the
+    bf16 tensor shaped like ``x``. Per-token FP8 is the only fused scheme today
+    (the consuming o_proj's a8w8 scheme); any other requested scheme asserts.
+
     ``gate`` may be strided (e.g. a column slice of a fused GEMM output): the
     kernel reads it via (outer, head) strides so no contiguous copy is needed.
     ``x`` is normed row-wise and is made contiguous (cheap; the caller's ``out``
     already is). Supports a 2D ``[M, H]`` or 3D ``[outer, heads, H]`` gate.
     """
+    if quant_type == QuantType.per_Token and quant_dtype == dtypes.fp8:
+        return _rmsnorm_gated_per_token_quant(x, weight, gate, eps, quant_dtype)
+    # Only the no-quant (bf16) path remains. Any other requested scheme is
+    # unsupported here -- fail loud rather than silently feed bf16 activations to
+    # a GEMM that expects quantized input.
+    assert quant_type in (None, QuantType.No), (
+        "rmsnorm_gated only fuses per-token FP8 quant; got "
+        f"quant_type={quant_type}, quant_dtype={quant_dtype}"
+    )
     h = x.shape[-1]
     x2 = x.reshape(-1, h)
     m = x2.shape[0]
@@ -156,6 +225,59 @@ def rmsnorm_gated(
     return y.reshape_as(x)
 
 
+def _rmsnorm_gated_per_token_quant(
+    x: torch.Tensor,
+    weight: torch.Tensor,
+    gate: torch.Tensor,
+    eps: float,
+    quant_dtype: torch.dtype,
+) -> tuple[torch.Tensor, torch.Tensor]:
+    """Per-head sigmoid-gated RMSNorm fused to per-token quant.
+
+    Same math as ``rmsnorm_gated`` (rmsnorm(x) over the last dim * weight *
+    sigmoid(gate)), but instead of a bf16 output it emits a ``quant_dtype`` tensor
+    plus one per-token scale (amax / dtype_max) over the flattened
+    ``heads * head_dim`` row, ready for o_proj's per-token a8w8 GEMM. ``gate`` may
+    be strided (a column slice of the fused in_proj output).
+
+    Returns ``(out [t, heads*head_dim], scale [t, 1] float32)``.
+    """
+    assert x.ndim == 3, f"expected [t, heads, head_dim], got {tuple(x.shape)}"
+    t, heads, H = x.shape
+    fp8_max = float(torch.finfo(quant_dtype).max)
+    if not _HAS_TRITON or t == 0 or H > 8192:
+        normed = _rmsnorm_gated_torch(x, weight, gate, eps).reshape(t, heads * H)
+        return get_hip_quant(QuantType.per_Token)(normed, quant_dtype=quant_dtype)
+    x = x.contiguous()
+    out = torch.empty((t, heads * H), dtype=quant_dtype, device=x.device)
+    scale = torch.empty((t, 1), dtype=torch.float32, device=x.device)
+    if gate.ndim == 3:
+        stride_g_outer, stride_g_head = gate.stride(0), gate.stride(1)
+    else:
+        # 2D [t, heads*H]: one logical head per row; head term drops out.
+        stride_g_outer, stride_g_head = gate.stride(0), 0
+    BLOCK = triton.next_power_of_2(H)
+    _rmsnorm_gated_fp8_per_token_kernel[(t,)](
+        x,
+        weight,
+        gate,
+        out,
+        scale,
+        H,
+        float(eps),
+        fp8_max,
+        x.stride(0),
+        x.stride(1),
+        stride_g_outer,
+        stride_g_head,
+        out.stride(0),
+        HEADS=heads,
+        HEADS_POW2=triton.next_power_of_2(heads),
+        BLOCK=BLOCK,
+    )
+    return out, scale
+
+
 # --------------------------------------------------------------------------- #
 # torch references (also the fallback when triton is unavailable)
 # --------------------------------------------------------------------------- #
diff --git a/atom/model_ops/linear.py b/atom/model_ops/linear.py
index 639b82d03b..5e975967ee 100644
--- a/atom/model_ops/linear.py
+++ b/atom/model_ops/linear.py
@@ -373,6 +373,24 @@ def gemm_a8w8_per_token_impl(
     )
 
 
+def _can_use_a8w8_preshuffle(output_size: int, input_size: int) -> bool:
+    """Whether an a8w8 weight can use the AITER bpreshuffle GEMM as-is.
+
+    ``shuffle_weight(..., layout=(16, 16))`` packs the output (N) dim in 16-row
+    tiles and the input (K) dim in ``BK = IK * 2 = 32``-col tiles (it asserts
+    ``x.shape[-1] % 32 == 0``). So N must be 16-aligned and K must be 32-aligned.
+    """
+    return output_size % 16 == 0 and input_size % 32 == 0
+
+
+def _a8w8_preshuffle_output_padding(output_size: int) -> int:
+    """Rows needed to pad an a8w8 weight's output dim (N) up to the GEMM's N-tile
+    (128). Returns 0 when already tile-aligned. Padding N to 128 also makes it
+    16-aligned, so the tuned preshuffle GEMM can run instead of falling back."""
+    remainder = output_size % 128
+    return 0 if remainder == 0 else 128 - remainder
+
+
 class LinearBase(nn.Module):
     def __init__(
         self,
@@ -487,6 +505,7 @@ def __init__(
             self.weight_scale.weight_loader = self.weight_loader
         self.need_normalize_e4m3fn_to_e4m3fnuz = params_dtype == torch.float8_e4m3fnuz
         self.quant_func = get_hip_quant(self.quant_type)
+        self.is_output_padded = False
 
     @staticmethod
     def weight_loader_process(
@@ -688,6 +707,8 @@ def online_quantize_weight(self):
         }
 
     def process_weights_after_loading(self):
+        if self.weight.numel() == 0:
+            return
         # Re-quantize before process_weights if online quantization is enabled
         if self.quant_config is not None and self.quant_config.online_quant:
             self.online_quantize_weight()
@@ -773,9 +794,9 @@ def process_weights_after_loading(self):
                     self, "needs_preshuffled_weight", False
                 ):
                     need_shuffle = True
-            if need_shuffle:
-                if self.weight.dim() == 2:
-                    shuffle_weights(self.weight)
+            if need_shuffle and self.weight.dim() == 2:
+                self.is_output_padded = self._maybe_pad_a8w8_preshuffle_output()
+                shuffle_weights(self.weight)
                 # self.weight_scale.data = fp4_utils.e8m0_shuffle(self.weight_scale.data)
         # shuffle weight scale once so no reshuffling for every gemm
         if self.quant_type == QuantType.per_1x32 and (
@@ -783,6 +804,42 @@ def process_weights_after_loading(self):
         ):
             self.weight_scale.data = fp4_utils.e8m0_shuffle(self.weight_scale.data)
 
+    def _maybe_pad_a8w8_preshuffle_output(self) -> bool:
+        if not (
+            self.quant_type == QuantType.per_Token and self.params_dtype == dtypes.fp8
+        ):
+            return False
+        if self.weight.dim() != 2:
+            return False
+        output_size, input_size = self.weight.shape
+        padding_size = _a8w8_preshuffle_output_padding(output_size)
+        if not _can_use_a8w8_preshuffle(output_size + padding_size, input_size):
+            # Padding the output (N) cannot make this weight preshuffle-able, i.e.
+            # the input dim K is not 32-aligned. Fail loudly here rather than let
+            # shuffle_weights hit its cryptic `x.shape[-1] % 32 == 0` assertion.
+            raise RuntimeError(
+                f"{self.prefix}: a8w8 bpreshuffle GEMM requires K % 32 == 0, got "
+                f"K={input_size}. Align K or run this layer via the triton a8w8 "
+                f"path (ATOM_USE_TRITON_GEMM=1)."
+            )
+        if padding_size == 0:
+            return False
+        self._output_size_before_padding = output_size
+        self.weight.data = torch.nn.functional.pad(
+            self.weight.data, (0, 0, 0, padding_size)
+        )
+        ws = self.weight_scale.data
+        self.weight_scale.data = torch.cat(
+            [ws, ws.new_ones((padding_size, *ws.shape[1:]))], dim=0
+        )
+        # Bias is also per-output-channel
+        if self.bias is not None:
+            b = self.bias.data
+            self.bias.data = torch.cat(
+                [b, b.new_zeros((padding_size, *b.shape[1:]))], dim=0
+            )
+        return True
+
     # linear mark trace shape/dtype helper
     def get_trace_prefix(
         self,
@@ -925,6 +982,9 @@ def forward(
                 )
                 if self.bias is not None:
                     y += self.bias
+        if self.is_output_padded:
+            # Drop the padded output rows
+            y = y[..., : self._output_size_before_padding]
         if self.tp_dim == 1 and self.tp_size > 1 and self.reduce_results:
             y = tensor_model_parallel_all_reduce(y)
         return y
diff --git a/atom/model_ops/module_dispatch_ops.py b/atom/model_ops/module_dispatch_ops.py
index a1141bb6ae..25e24fae6d 100644
--- a/atom/model_ops/module_dispatch_ops.py
+++ b/atom/model_ops/module_dispatch_ops.py
@@ -19,18 +19,19 @@
 the methods listed in each op's docstring.
 
 Currently registered:
-  - torch.ops.aiter.maybe_dual_stream_forward — V2/V3.2/V4 MoE
+  - torch.ops.aiter.maybe_dual_stream_forward — V2/V3.2/V4/K3 MoE
   - torch.ops.aiter.indexer_score_topk       — V4 sparse indexer
 """
 
 import torch
 
-from atom.config import get_current_atom_config
+from atom.config import CUDAGraphMode, get_current_atom_config
 from atom.utils import envs
 from atom.utils.custom_register import direct_register_custom_op
+from atom.utils.forward_context import get_current_cudagraph_runtime_mode
 
 # ---------------------------------------------------------------------------
-# Dual-stream MoE dispatch (V2 / V3.2 / V4)
+# Dual-stream MoE dispatch (V2 / V3.2 / V4 / K3)
 # ---------------------------------------------------------------------------
 #
 # Caller contract (the MoE module looked up by `layer_name`):
@@ -54,13 +55,12 @@ def maybe_dual_stream_forward(
     # Under TBO the two micro-batches already overlap on separate threads
     from atom.utils.tbo.ubatching import tbo_active
 
-    # PIECEWISE cudagraph only: dual_stream_moe_forward forks work onto
-    # `alt_stream` and does a caching-allocator alloc there; under PIECEWISE
-    # per-piece capture close the dual stream
-    compilation_config = get_current_atom_config().compilation_config
-    cudagraph_mode = getattr(compilation_config, "cudagraph_mode", None)
+    # Graph ownership belongs to the active frontend.  Only a concrete
+    # PIECEWISE runtime decision is unsafe here: per-piece capture closes over
+    # the main stream while this forward forks work onto `alt_stream`.  Eager
+    # NONE and whole-model FULL capture both support the fork/join topology.
     is_piecewise_cudagraph = (
-        cudagraph_mode is not None and cudagraph_mode.requires_piecewise_compilation()
+        get_current_cudagraph_runtime_mode() == CUDAGraphMode.PIECEWISE
     )
 
     if (
diff --git a/atom/models/kimi_k3.py b/atom/models/kimi_k3.py
index eb9375e3ed..2732451d75 100644
--- a/atom/models/kimi_k3.py
+++ b/atom/models/kimi_k3.py
@@ -11,18 +11,22 @@
 from typing import ClassVar
 
 import torch
-from aiter import ActivationType, QuantType, fused_qk_rmsnorm
+from aiter import ActivationType, QuantType, dtypes
 from aiter.dist.communication_op import tensor_model_parallel_all_reduce
 from aiter.dist.parallel_state import (
     get_pp_group,
     get_tensor_model_parallel_rank,
     get_tensor_model_parallel_world_size,
 )
-from aiter.jit.utils.torch_guard import torch_compile_guard
 from einops import rearrange
 from torch import nn
 
 from atom.config import Config, QuantizationConfig, get_current_atom_config
+
+# Side-effect import: registers `torch.ops.aiter.maybe_dual_stream_forward`, the
+# Dynamo-opaque custom op that dispatches the MoE between single- and dual-stream
+# forwards (shared with deepseek_v2/v4). Imported for the registration only.
+from atom.model_ops import module_dispatch_ops as _module_dispatch_ops  # noqa: F401
 from atom.model_ops.attention_mla import MLAModules
 from atom.model_ops.base_attention import Attention
 from atom.model_ops.embed_head import ParallelLMHead, VocabParallelEmbedding
@@ -36,6 +40,8 @@
     MergedReplicatedLinear,
     ReplicatedLinear,
     RowParallelLinear,
+    use_fp4_non_shuffle_triton_gemm,
+    use_triton_gemm,
 )
 from atom.model_ops.mamba_ops.causal_conv1d import (
     causal_conv1d_fn,
@@ -51,7 +57,8 @@
     make_layers,
     maybe_prefix,
 )
-from atom.utils import mark_spliting_op
+from atom.quant_spec import should_skip_online_quant
+from atom.utils import envs, mark_spliting_op
 from atom.utils.decorators import support_torch_compile
 from atom.utils.forward_context import get_forward_context
 
@@ -134,39 +141,37 @@ def _extract_layer_idx(prefix: str) -> int:
     return 0
 
 
-def _fused_qk_rmsnorm_fake(
-    q: torch.Tensor,
-    q_weight: torch.Tensor,
-    q_eps: float,
-    k: torch.Tensor,
-    k_weight: torch.Tensor,
-    k_eps: float,
-) -> tuple[torch.Tensor, torch.Tensor]:
-    return q.new_empty(q.shape), k.new_empty(k.shape)
+# RMSNorm+quant fusion is scheme-agnostic: the aiter fused RMSNorm kernels
+# (RMSNorm._aiter_rms_quant and deepseek's _fuse_rmsnorm_quant) emit any of these
+# dynamic activation quant layouts, so a preceding norm can fold the quant for a
+# Linear that runs one of them.
+_RMS_FUSABLE_QUANT_TYPES = (
+    QuantType.per_1x32,
+    QuantType.per_1x128,
+    QuantType.per_Token,
+)
 
 
-@torch_compile_guard(gen_fake=_fused_qk_rmsnorm_fake, mutates_args=[])
-def _fused_qk_rmsnorm(
-    q: torch.Tensor,
-    q_weight: torch.Tensor,
-    q_eps: float,
-    k: torch.Tensor,
-    k_weight: torch.Tensor,
-    k_eps: float,
-) -> tuple[torch.Tensor, torch.Tensor]:
-    q_out = torch.empty(q.shape, dtype=q.dtype, device=q.device)
-    k_out = torch.empty(k.shape, dtype=k.dtype, device=k.device)
-    fused_qk_rmsnorm(
-        q_out_quantized=q_out,
-        q=q,
-        q_weight=q_weight,
-        q_epsilon=q_eps,
-        k_out=k_out,
-        k=k,
-        k_weight=k_weight,
-        k_epsilon=k_eps,
-    )
-    return q_out, k_out
+def _effective_layer_quant(
+    quant_config: QuantizationConfig | None, prefix: str
+) -> tuple[QuantType, torch.dtype | None]:
+    """Resolve the ``(quant_type, quant_dtype)`` a Linear runs with at runtime.
+
+    Same resolution the Linear itself performs (mirrors
+    ``LinearBase.online_quantize_weight`` and deepseek_v2's MLA setup): the static
+    checkpoint scheme, overridden by the online-quant target when that override
+    actually applies (``should_skip_online_quant``). A preceding RMSNorm uses this
+    to decide whether -- and in which scheme (fp8 / fp4x2, per-token / block) -- to
+    fuse its activation quant, rather than hard-coding one layout.
+    """
+    if quant_config is None:
+        return QuantType.No, None
+    cfg = quant_config.get_layer_quant_config(prefix)
+    if quant_config.online_quant:
+        online_cfg = quant_config.get_layer_quant_config(prefix, use_online_quant=True)
+        if not should_skip_online_quant(cfg.quant_type, cfg.quant_dtype, online_cfg):
+            cfg = online_cfg
+    return cfg.quant_type, cfg.quant_dtype
 
 
 class _NoPositionalRotaryEmbedding(RotaryEmbedding):
@@ -204,15 +209,33 @@ def forward(self, x: torch.Tensor) -> torch.Tensor:
 
 
 class KimiRMSNormGated(nn.Module):
-    def __init__(self, hidden_size: int, eps: float):
+    def __init__(
+        self,
+        hidden_size: int,
+        eps: float,
+        quant_type: QuantType | None = None,
+        quant_dtype: torch.dtype | None = None,
+    ):
         super().__init__()
         self.weight = atom_parameter(torch.ones(hidden_size))
         self.variance_epsilon = eps
+        # When ``quant_type`` names a fusable per-token scheme, the per-head
+        # sigmoid-gated norm also emits (quantized, scale) so the consuming
+        # o_proj skips its standalone quant; otherwise it returns a bf16 tensor.
+        self.quant_type = quant_type
+        self.quant_dtype = quant_dtype
 
-    def forward(self, x: torch.Tensor, gate: torch.Tensor) -> torch.Tensor:
+    def forward(self, x: torch.Tensor, gate: torch.Tensor):
         from atom.model_ops.kimi_k3 import rmsnorm_gated
 
-        return rmsnorm_gated(x, self.weight, gate, self.variance_epsilon)
+        return rmsnorm_gated(
+            x,
+            self.weight,
+            gate,
+            self.variance_epsilon,
+            quant_type=self.quant_type,
+            quant_dtype=self.quant_dtype,
+        )
 
 
 def _sharded_vector_loader(tp_rank: int, tp_size: int):
@@ -259,7 +282,12 @@ def __init__(
         )
 
     def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return self.down_proj(self.act_fn(self.gate_up_proj(x)))
+        # `x` arrives as a (fp8, scale) tuple when the preceding RMSNorm fused its
+        # activation quant (dense-MLP layers)
+        x_scale = None
+        if isinstance(x, tuple):
+            x, x_scale = x
+        return self.down_proj(self.act_fn(self.gate_up_proj(x, x_scale)))
 
 
 class KimiSparseMoeBlock(nn.Module):
@@ -268,10 +296,12 @@ def __init__(
         config,
         quant_config: QuantizationConfig | None = None,
         prefix: str = "",
+        alt_stream: torch.cuda.Stream | None = None,
     ):
         super().__init__()
         self.config = config
         self.prefix = prefix
+        self.alt_stream = alt_stream
         self.hidden_dim = config.hidden_size
         self.num_experts = config.num_experts
         self.top_k = config.num_experts_per_token
@@ -359,17 +389,64 @@ def _routed_source_quant_dtype(layer_prefix: str) -> torch.dtype | None:
                 source_quant_dtype=_routed_source_quant_dtype(up_proj_prefix),
                 prefix=up_proj_prefix,
             )
+            up_proj_quant_type, up_proj_quant_dtype = _effective_layer_quant(
+                quant_config, up_proj_prefix
+            )
+            latent_moe_use_norm = getattr(config, "latent_moe_use_norm", False)
+            # AITER RMSNorm+quant emits the activation layout consumed directly by
+            # the routed up-projection. FP4 Triton paths choose an M-dependent
+            # shuffled/non-shuffled scale layout, so keep those on their existing
+            # standalone quant path until the fused kernel supports both layouts.
+            fp4_triton_active = up_proj_quant_type == QuantType.per_1x32 and (
+                use_triton_gemm() or use_fp4_non_shuffle_triton_gemm()
+            )
+            self.fuse_routed_norm_quant = latent_moe_use_norm and (
+                (
+                    up_proj_quant_type == QuantType.per_1x32
+                    and up_proj_quant_dtype == dtypes.fp4x2
+                    and not fp4_triton_active
+                )
+                or (
+                    up_proj_quant_type in (QuantType.per_1x128, QuantType.per_Token)
+                    and up_proj_quant_dtype == dtypes.fp8
+                )
+            )
             self.routed_expert_norm = (
-                RMSNorm(self.moe_hidden_size, eps=config.rms_norm_eps)
-                if getattr(config, "latent_moe_use_norm", False)
+                RMSNorm(
+                    self.moe_hidden_size,
+                    eps=config.rms_norm_eps,
+                    fused_quant=self.fuse_routed_norm_quant,
+                    quant_config=quant_config,
+                    prefix=up_proj_prefix,
+                )
+                if latent_moe_use_norm
                 else None
             )
 
-    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
-        return self._forward_impl(hidden_states)
+        # Dual-stream gate: overlap the shared-expert GEMMs (on alt_stream) with
+        # the routed-expert path (on the main stream). Only meaningful when a
+        # shared branch exists and an alt_stream was threaded in. TBO already
+        # provides its own overlap, so the two are mutually exclusive.
+        self._use_dual_stream = False
+        if self.shared_experts is not None and self.alt_stream is not None:
+            tbo_active = get_current_atom_config().enable_tbo
+            if envs.ATOM_DUAL_STREAM_MOE_TOKEN_THRESHOLD > 0 and not tbo_active:
+                self._use_dual_stream = True
+        if self._use_dual_stream:
+            # Register self so `maybe_dual_stream_forward` can look this module up
+            # by prefix from static_forward_context (the op is Dynamo-opaque).
+            cc = get_current_atom_config().compilation_config
+            cc.static_forward_context[self.prefix] = self
 
-    def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
-        identity = hidden_states
+    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        if self._use_dual_stream:
+            return torch.ops.aiter.maybe_dual_stream_forward(hidden_states, self.prefix)
+        return self.single_stream_moe_forward(hidden_states)
+
+    def routed_expert_forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        """Routed-expert path only. For the latent MoE this includes the routed
+        all-reduce (required before the nonlinear routed_expert_norm); the shared
+        branch is handled by the caller."""
         router_logits = self.gate(hidden_states)
         routed_input = (
             self.routed_expert_down_proj(hidden_states)
@@ -387,7 +464,19 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
                 routed_output = tensor_model_parallel_all_reduce(routed_output)
             if self.routed_expert_norm is not None:
                 routed_output = self.routed_expert_norm(routed_output)
-            routed_output = self.routed_expert_up_proj(routed_output)
+            if isinstance(routed_output, tuple):
+                routed_output, routed_output_scale = routed_output
+                routed_output = self.routed_expert_up_proj(
+                    routed_output, x_scale=routed_output_scale
+                )
+            else:
+                routed_output = self.routed_expert_up_proj(routed_output)
+        return routed_output
+
+    def single_stream_moe_forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        identity = hidden_states
+        routed_output = self.routed_expert_forward(hidden_states)
+        if self.use_latent_moe:
             if self.shared_experts is not None:
                 # Shared branch is TP-partial (down_proj is row-parallel); reduce
                 # it separately and add to the already-full routed output.
@@ -405,6 +494,55 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
             routed_output = tensor_model_parallel_all_reduce(routed_output)
         return routed_output
 
+    def dual_stream_moe_forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        # Queue routed pre-AR work first on the current stream, then run the
+        # shared-expert path on alt_stream. The latent path keeps both all-reduces
+        # on their respective streams while preserving shared AR -> routed AR
+        # order on the single TP communicator.
+        current = torch.cuda.current_stream()
+        alt = self.alt_stream
+        alt.wait_stream(current)
+
+        if self.use_latent_moe:
+            router_logits = self.gate(hidden_states)
+            routed_input = self.routed_expert_down_proj(hidden_states)
+            routed_output = self.experts(routed_input, router_logits)
+        else:
+            routed_output = self.routed_expert_forward(hidden_states)
+
+        with torch.cuda.stream(alt):
+            shared_output = self.shared_experts(hidden_states)
+            if self.use_latent_moe and self.tp_size > 1:
+                shared_output = tensor_model_parallel_all_reduce(shared_output)
+
+        if self.use_latent_moe:
+            if self.tp_size > 1:
+                current.wait_stream(alt)
+                routed_output = tensor_model_parallel_all_reduce(routed_output)
+
+            if self.routed_expert_norm is not None:
+                routed_output = self.routed_expert_norm(routed_output)
+            if isinstance(routed_output, tuple):
+                routed_output, routed_output_scale = routed_output
+                routed_output = self.routed_expert_up_proj(
+                    routed_output,
+                    x_scale=routed_output_scale,
+                )
+            else:
+                routed_output = self.routed_expert_up_proj(routed_output)
+
+            if self.tp_size == 1:
+                current.wait_stream(alt)
+            shared_output.record_stream(current)
+            return routed_output + shared_output
+
+        # Non-latent: shared has no AR yet; single deferred AR over the sum.
+        current.wait_stream(alt)
+        routed_output = routed_output + shared_output
+        if self.tp_size > 1:
+            routed_output = tensor_model_parallel_all_reduce(routed_output)
+        return routed_output
+
 
 class KimiFullAttention(nn.Module):
     def __init__(
@@ -468,12 +606,15 @@ def __init__(
 
         rope_parameters = getattr(config, "rope_parameters", None) or {}
         rope_theta = rope_parameters.get("rope_theta") or 10000.0
+        # max_position_embeddings field only exists in the text config
+        _text_max_pos = getattr(config, "max_position_embeddings", None)
+        rope_max_position = int(
+            _text_max_pos or getattr(atom_config, "max_model_len", None) or 16384
+        )
         self.rotary_emb = _NoPositionalRotaryEmbedding(
             head_size=self.qk_rope_head_dim,
             rotary_dim=self.qk_rope_head_dim,
-            max_position_embeddings=int(
-                getattr(atom_config, "max_model_len", None) or 16384
-            ),
+            max_position_embeddings=rope_max_position,
             base=rope_theta,
         )
         mla_modules = MLAModules(
@@ -504,31 +645,86 @@ def __init__(
             prefix=prefix,
         )
 
+        qknorm_type, qknorm_dtype = _effective_layer_quant(
+            quant_config, f"{prefix}.q_b_proj"
+        )
+        self.fuse_qknorm_quant = qknorm_dtype in (dtypes.fp8, dtypes.fp4x2)
+        self.qknorm_dtype = qknorm_dtype if self.fuse_qknorm_quant else torch.bfloat16
+        self.qknorm_quant_type_value = (
+            qknorm_type.value if self.fuse_qknorm_quant else QuantType.No.value
+        )
+        # input_layernorm fuses its activation quant only when BOTH consumers of
+        # the normed hidden state -- fused_qkv_a_proj and g_proj -- run with the
+        # same fusable RMSNorm quant scheme (else a mismatched consumer mis-GEMMs).
+        a_scheme = _effective_layer_quant(quant_config, f"{prefix}.fused_qkv_a_proj")
+        g_scheme = _effective_layer_quant(quant_config, f"{prefix}.g_proj")
+        self.fuse_input_norm_quant = (
+            a_scheme[0] in _RMS_FUSABLE_QUANT_TYPES and a_scheme == g_scheme
+        )
+        self.input_quant_prefix = f"{prefix}.fused_qkv_a_proj"
+
     def forward(
         self, positions: torch.Tensor, hidden_states: torch.Tensor
     ) -> torch.Tensor:
-        q, kv, k_rope = torch.split(
-            self.fused_qkv_a_proj(hidden_states),
+        # deepseek_v2 pattern: one _fuse_rmsnorm_quant kernel does q_a norm +
+        # kv_a norm (+ q-activation quant), then q's scale is forwarded into the
+        # MLA module (q_proj consumes it).
+        from atom.models.deepseek_v2 import _fuse_rmsnorm_quant
+
+        # hidden_states is a (fp8, scale) tuple when input_layernorm fused the
+        # quant; both fused_qkv_a_proj and g_proj consume it directly.
+        hidden_states_scale = None
+        if isinstance(hidden_states, tuple):
+            hidden_states, hidden_states_scale = hidden_states
+
+        q_c, kv_c, k_rope = torch.split(
+            self.fused_qkv_a_proj(hidden_states, hidden_states_scale),
             [self.q_lora_rank, self.kv_lora_rank, self.qk_rope_head_dim],
             dim=-1,
         )
-        q, kv = _fused_qk_rmsnorm(
-            q,
+        q_shuffle = False
+        q_scale_shuffle_padding = False
+        if self.qknorm_dtype == dtypes.fp4x2:
+            from atom.model_ops.linear import use_triton_gemm
+            from atom.models.deepseek_v2 import _mxfp4_activation_quant_layout
+
+            if not use_triton_gemm():
+                q_shuffle, q_scale_shuffle_padding = _mxfp4_activation_quant_layout(
+                    q_c.shape[0]
+                )
+        (q, q_scale), _, kv, _ = _fuse_rmsnorm_quant(
+            q_c,
             self.q_a_layernorm.weight,
             self.q_a_layernorm.eps,
-            kv,
+            kv_c,
             self.kv_a_layernorm.weight,
             self.kv_a_layernorm.eps,
+            None,
+            dtype_quant=self.qknorm_dtype,
+            shuffle=q_shuffle,
+            scale_shuffle_padding=q_scale_shuffle_padding,
+            group_size=128,
+            quant_type=self.qknorm_quant_type_value,
+            output_unquantized_inp1=False,
+            transpose_scale=True,
+        )
+        attn_out = self.attn(q, kv, k_rope, positions, q_scale=q_scale)
+        attn_out = attn_out * torch.sigmoid(
+            self.g_proj(hidden_states, hidden_states_scale)
         )
-        attn_out = self.attn(q, kv, k_rope, positions)
-        attn_out = attn_out * torch.sigmoid(self.g_proj(hidden_states))
         return self.o_proj(attn_out)
 
 
 def _kda_attention_with_output_fake(
-    hidden_states: torch.Tensor, layer_name: str
+    hidden_states: torch.Tensor,
+    hidden_states_scale: torch.Tensor | None,
+    layer_name: str,
 ) -> torch.Tensor:
-    return torch.empty_like(hidden_states)
+    # The mixer output (o_proj) is always bf16 even when the input activation is
+    # fp8 (fused input_layernorm+quant), so pin the dtype rather than empty_like.
+    return torch.empty(
+        hidden_states.shape, dtype=torch.bfloat16, device=hidden_states.device
+    )
 
 
 @mark_spliting_op(
@@ -537,7 +733,9 @@ def _kda_attention_with_output_fake(
     mutates_args=[],
 )
 def kda_attention_with_output(
-    hidden_states: torch.Tensor, layer_name: str
+    hidden_states: torch.Tensor,
+    hidden_states_scale: torch.Tensor | None,
+    layer_name: str,
 ) -> torch.Tensor:
     """Opaque splitting-op boundary for the KDA mixer.
 
@@ -551,7 +749,7 @@ def kda_attention_with_output(
     self = get_current_atom_config().compilation_config.static_forward_context[
         layer_name
     ]
-    return self._forward_impl(hidden_states)
+    return self._forward_impl(hidden_states, hidden_states_scale)
 
 
 class KimiKDAAttention(nn.Module):
@@ -666,7 +864,13 @@ def __init__(
             quant_config=quant_config,
             prefix=f"{prefix}.f_b_proj",
         )
-        self.o_norm = KimiRMSNormGated(self.head_dim, eps=config.rms_norm_eps)
+        o_type, o_dtype = _effective_layer_quant(quant_config, f"{prefix}.o_proj")
+        self.o_norm = KimiRMSNormGated(
+            self.head_dim,
+            eps=config.rms_norm_eps,
+            quant_type=o_type,
+            quant_dtype=o_dtype,
+        )
         self.o_proj = RowParallelLinear(
             self.proj_size,
             self.hidden_size,
@@ -675,6 +879,14 @@ def __init__(
             prefix=f"{prefix}.o_proj",
         )
 
+        # The decoder's input_layernorm can fuse its activation quant into the
+        # single fused in_proj GEMM (q|k|v|g|b|f_a) that consumes the normed hidden
+        # state; the (fp8, scale) rides the splitting custom op into _forward_impl.
+        # Enabled for any fusable RMSNorm quant scheme (fp8 / fp4x2).
+        in_proj_type, _ = _effective_layer_quant(quant_config, f"{prefix}.in_proj")
+        self.fuse_input_norm_quant = in_proj_type in _RMS_FUSABLE_QUANT_TYPES
+        self.input_quant_prefix = f"{prefix}.in_proj"
+
     def process_weights_after_loading(self) -> None:
         """Fuse all hidden-input projections into the single in-proj (one GEMM).
 
@@ -778,15 +990,30 @@ def _run_kda(
         return chunk_kda(**kwargs, disable_recompute=True)
 
     def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        # hidden_states is a (fp8, scale) tuple when input_layernorm fused the
+        # per-token quant; carry the scale through the opaque splitting custom op
+        # so in_proj consumes it in _forward_impl.
+        hidden_states_scale = None
+        if isinstance(hidden_states, tuple):
+            hidden_states, hidden_states_scale = hidden_states
         # Route through the opaque custom op so torch.compile splits the graph
         # here instead of tracing the stateful recurrence in _forward_impl.
-        return torch.ops.aiter.kda_attention_with_output(hidden_states, self.layer_name)
+        return torch.ops.aiter.kda_attention_with_output(
+            hidden_states, hidden_states_scale, self.layer_name
+        )
 
-    def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
+    def _forward_impl(
+        self,
+        hidden_states: torch.Tensor,
+        hidden_states_scale: torch.Tensor | None = None,
+    ) -> torch.Tensor:
         fwd_ctx = get_forward_context()
         gdn_metadata = getattr(fwd_ctx.attn_metadata, "gdn_metadata", None)
         if gdn_metadata is None:
-            return hidden_states.new_zeros(hidden_states.shape)
+            # Output is bf16 even when the input activation is fp8 (fused quant).
+            return torch.zeros(
+                hidden_states.shape, dtype=torch.bfloat16, device=hidden_states.device
+            )
 
         cache = fwd_ctx.kv_cache_data[f"layer_{self.layer_num}"]
         conv_state = cache.k_cache
@@ -796,6 +1023,8 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
 
         num_actual_tokens = gdn_metadata.num_actual_tokens
         hidden_states = hidden_states[:num_actual_tokens]
+        if hidden_states_scale is not None:
+            hidden_states_scale = hidden_states_scale[:num_actual_tokens]
         # Single fused in-proj GEMM producing [q | k | v | g]; slice out each
         # part. `out_gate` is the KDA output gate consumed at o_norm below
         # (computed here so it rides the same GEMM instead of a separate one
@@ -806,7 +1035,7 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
         lp = self.local_proj_size
         nlh = self.num_local_heads
         hd = self.head_dim
-        fused_in = self.in_proj(hidden_states)
+        fused_in = self.in_proj(hidden_states, x_scale=hidden_states_scale)
         # No .contiguous() needed: mixed_qkv is a column slice (feature stride 1,
         # row stride N_fused). Both causal-conv consumers read the token stride
         # from the tensor itself — causal_conv1d_fn uses x.stride(1) after
@@ -822,7 +1051,8 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
         f_a = fused_in[..., 4 * lp + nlh : 4 * lp + nlh + hd].contiguous()
         gate = self.f_b_proj(f_a)
         gate = rearrange(gate, "t (h d) -> 1 t h d", d=self.head_dim)
-        out = hidden_states.new_empty(
+        # Allocate from fused_in (bf16), not hidden_states, which may be fp8.
+        out = fused_in.new_empty(
             (num_actual_tokens, self.num_local_heads, self.head_dim)
         )
 
@@ -963,8 +1193,15 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
         else:
             out.zero_()
 
-        out = self.o_norm(out, rearrange(out_gate, "t (h d) -> t h d", d=self.head_dim))
-        return self.o_proj(rearrange(out, "t h d -> t (h d)"))
+        normed = self.o_norm(
+            out, rearrange(out_gate, "t (h d) -> t h d", d=self.head_dim)
+        )
+        # A fused per-token quant makes o_norm return (quantized, scale); feed it
+        # straight to o_proj's x_scale path. Otherwise it is a bf16 tensor.
+        if isinstance(normed, tuple):
+            o_fp8, o_scale = normed
+            return self.o_proj(o_fp8, x_scale=o_scale)
+        return self.o_proj(rearrange(normed, "t h d -> t (h d)"))
 
 
 class KimiDecoderLayer(nn.Module):
@@ -973,6 +1210,7 @@ def __init__(
         atom_config: Config,
         prefix: str,
         layer_num: int = 0,
+        alt_stream: torch.cuda.Stream | None = None,
     ):
         super().__init__()
         config = _text_config(atom_config.hf_config)
@@ -1000,14 +1238,42 @@ def __init__(
                 config,
                 quant_config=quant_config,
                 prefix=f"{prefix}.block_sparse_moe",
+                alt_stream=alt_stream,
             )
         else:
             self.mlp = KimiMLP(
                 config, quant_config=quant_config, prefix=f"{prefix}.mlp"
             )
-        self.input_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
+        # Fuse the activation quant into input_layernorm when the attention input
+        # projection(s) run with a fusable quant scheme (self_attn decides and
+        # exposes the flag + representative prefix). The normed output then flows
+        # to the attention as a (fp8, scale) tuple instead of a bf16 tensor + a
+        # standalone quant op.
+        self.input_layernorm = RMSNorm(
+            config.hidden_size,
+            eps=config.rms_norm_eps,
+            fused_quant=self.self_attn.fuse_input_norm_quant,
+            quant_config=(
+                quant_config if self.self_attn.fuse_input_norm_quant else None
+            ),
+            prefix=self.self_attn.input_quant_prefix,
+        )
+        # Fuse post_attention_layernorm's quant into the dense-MLP gate_up_proj.
+        # MoE layers are skipped: their router gate is unquantized and the routed
+        # experts are excluded, so the normed output has mixed-precision consumers.
+        if hasattr(self, "mlp"):
+            ffn_type, _ = _effective_layer_quant(
+                quant_config, f"{prefix}.mlp.gate_up_proj"
+            )
+            self.fuse_ffn_norm_quant = ffn_type in _RMS_FUSABLE_QUANT_TYPES
+        else:
+            self.fuse_ffn_norm_quant = False
         self.post_attention_layernorm = RMSNorm(
-            config.hidden_size, eps=config.rms_norm_eps
+            config.hidden_size,
+            eps=config.rms_norm_eps,
+            fused_quant=self.fuse_ffn_norm_quant,
+            quant_config=quant_config if self.fuse_ffn_norm_quant else None,
+            prefix=f"{prefix}.mlp.gate_up_proj",
         )
 
         self.use_attn_residuals = (
@@ -1163,12 +1429,21 @@ def __init__(self, atom_config: Config, prefix: str = ""):
         else:
             self.embed_tokens = PPMissingLayer()
 
+        # Shared second stream for dual-stream MoE (shared-expert GEMMs overlap the
+        # routed path). Created once and threaded into every decoder layer; only
+        # used when the model has shared experts.
+        self.alt_stream = None
+        if getattr(config, "num_shared_experts", 0):
+            self.alt_stream = torch.cuda.Stream()
+        _alt_stream = self.alt_stream
+
         self.start_layer, self.end_layer, self.layers = make_layers(
             config.num_hidden_layers,
             lambda prefix, layer_num=None: KimiDecoderLayer(
                 atom_config,
                 prefix=prefix,
                 layer_num=layer_num or 0,
+                alt_stream=_alt_stream,
             ),
             prefix=f"{prefix}.layers",
             layer_num_offset=0,
diff --git a/atom/plugin/vllm/models/kimi_k3.py b/atom/plugin/vllm/models/kimi_k3.py
index 9c26c6a1c7..3e4e3c60a0 100644
--- a/atom/plugin/vllm/models/kimi_k3.py
+++ b/atom/plugin/vllm/models/kimi_k3.py
@@ -108,11 +108,17 @@ def get_state_shape(self) -> tuple[tuple[int, ...], tuple[int, ...]]:
     def mamba_type(self) -> MambaAttentionBackendEnum:
         return MambaAttentionBackendEnum.GDN_ATTN
 
-    def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
+    def _forward_impl(
+        self,
+        hidden_states: torch.Tensor,
+        hidden_states_scale: torch.Tensor | None = None,
+    ) -> torch.Tensor:
         vllm_context = get_vllm_forward_context()
         attn_metadata = vllm_context.attn_metadata
         if attn_metadata is None:
-            return hidden_states.new_zeros(hidden_states.shape)
+            return torch.zeros(
+                hidden_states.shape, dtype=torch.bfloat16, device=hidden_states.device
+            )
 
         if not isinstance(attn_metadata, dict):
             raise TypeError("Kimi-K3 vLLM attention metadata must be layer-indexed")
@@ -135,7 +141,7 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
         atom_context.attn_metadata = self._atom_metadata
         atom_context.kv_cache_data = self._atom_kv_cache_data
         try:
-            output = super()._forward_impl(hidden_states)
+            output = super()._forward_impl(hidden_states, hidden_states_scale)
         finally:
             atom_context.attn_metadata = previous_metadata
             atom_context.kv_cache_data = previous_kv_cache_data
@@ -143,7 +149,7 @@ def _forward_impl(self, hidden_states: torch.Tensor) -> torch.Tensor:
         # vLLM pads token rows to the selected piecewise/full graph bucket,
         # while GDN metadata tracks only real tokens. The native KDA path
         # intentionally slices to num_actual_tokens; restore the graph bucket
-        # width so this custom op matches its empty_like fake implementation.
+        # width so this custom op matches its fake implementation's output shape.
         if output.shape[0] < hidden_states.shape[0]:
             output = torch.nn.functional.pad(
                 output,
diff --git a/atom/utils/forward_context.py b/atom/utils/forward_context.py
index b5126f6398..18c9596ff9 100644
--- a/atom/utils/forward_context.py
+++ b/atom/utils/forward_context.py
@@ -11,7 +11,7 @@
 import numpy as np
 import torch
 
-from atom.config import Config, KVCacheTensor, ParallelConfig
+from atom.config import Config, CUDAGraphMode, KVCacheTensor, ParallelConfig
 
 
 class AttnState(Enum):
@@ -600,6 +600,57 @@ def get_forward_context() -> ForwardContext:
     return _forward_context
 
 
+def _normalize_cudagraph_runtime_mode(mode: Any) -> CUDAGraphMode | None:
+    """Normalize a frontend runtime mode to ATOM's concrete enum.
+
+    Frontends own their graph dispatch and therefore use distinct enum
+    classes.  Match by name rather than value so their enum layouts can evolve
+    independently.  Composite configuration modes are deliberately rejected:
+    a forward context must describe the concrete NONE/PIECEWISE/FULL decision
+    for the current batch.
+    """
+    name = mode if isinstance(mode, str) else getattr(mode, "name", None)
+    if name not in {"NONE", "PIECEWISE", "FULL"}:
+        return None
+    return CUDAGraphMode[name]
+
+
+def get_current_cudagraph_runtime_mode() -> CUDAGraphMode:
+    """Return the concrete graph mode for the active model forward.
+
+    In vLLM plugin mode graph capture/replay is owned by vLLM, so its forward
+    context is authoritative.  Native ATOM records the same decision on its
+    own ForwardContext.  An unavailable/unknown context is treated as NONE:
+    eager dual-stream execution is valid, and some vLLM runners expose NONE
+    while a whole-model FULL graph is being captured.  Replay does not execute
+    this Python dispatcher.
+    """
+    from atom.plugin import is_vllm
+
+    if is_vllm():
+        try:
+            from vllm.forward_context import (
+                get_forward_context as get_vllm_forward_context,
+            )
+            from vllm.forward_context import (
+                is_forward_context_available,
+            )
+
+            if is_forward_context_available():
+                mode = _normalize_cudagraph_runtime_mode(
+                    get_vllm_forward_context().cudagraph_runtime_mode
+                )
+                if mode is not None:
+                    return mode
+        except (ImportError, AttributeError, AssertionError):
+            pass
+
+    mode = _normalize_cudagraph_runtime_mode(
+        getattr(get_forward_context(), "cudagraph_runtime_mode", None)
+    )
+    return mode if mode is not None else CUDAGraphMode.NONE
+
+
 def set_forward_context(
     attn_metadata: AttentionMetaData,
     atom_config: Config,
diff --git a/recipes/Kimi-K3.md b/recipes/Kimi-K3.md
index cf59c53b6f..f2cd10084a 100644
--- a/recipes/Kimi-K3.md
+++ b/recipes/Kimi-K3.md
@@ -30,7 +30,8 @@ python -m atom.entrypoints.openai_server \
   --max-num-batched-tokens 16384 \
   --gpu-memory-utilization 0.93 \
   --block-size 128 \
-  --no-enable_prefix_caching
+  --no-enable_prefix_caching \
+  --online_quant_config '{"global_quant_config": "ptpc_fp8", "exclude_layer": ["lm_head", "model.embed_tokens", "*self_attn.[qkv]_conv1d*", "*block_sparse_moe.experts*", "*block_sparse_moe.routed_expert_*", "*vision_tower*", "*mm_projector*"]}'
 ```
 
 Kimi full-attention layers use true MLA with a compressed latent KV cache. Aiter MLA is selected by default; `ATOM_USE_TRITON_MLA=1` selects the Triton MLA implementation when that configuration has been validated.
diff --git a/recipes/atom_vllm/Kimi-K3.md b/recipes/atom_vllm/Kimi-K3.md
index 243cb27019..4931f388bd 100644
--- a/recipes/atom_vllm/Kimi-K3.md
+++ b/recipes/atom_vllm/Kimi-K3.md
@@ -28,18 +28,6 @@ pip install -e /path/to/ATOM --no-deps
 ```bash
 MODEL=/path/to/Kimi-K3
 
-export AITER_LOG_LEVEL=WARNING
-export ATOM_LOADER_USE_THREADPOOL=1
-export ATOM_LOADER_THREADPOOL_WORKERS=16
-export ATOM_SYNC_AFTER_LOAD=1
-export ATOM_DIST_TIMEOUT_SECONDS=3600
-
-export ATOM_USE_TRITON_GEMM=1
-export AITER_USE_GROUPED_GEMM=0
-export ATOM_USE_TRITON_MOE=0
-export AITER_FLYDSL_FORCE=1
-export AITER_FORCE_GFX1250=0
-
 vllm serve "${MODEL}" \
     --host 0.0.0.0 \
     --port 8000 \
@@ -53,8 +41,8 @@ vllm serve "${MODEL}" \
     --gpu-memory-utilization 0.93 \
     --block-size 128 \
     --no-enable-prefix-caching \
-    --no-async-scheduling \
-    --compilation-config '{"cudagraph_mode":"FULL_AND_PIECEWISE"}'
+    --compilation-config '{"cudagraph_mode":"FULL_AND_PIECEWISE"}' \
+    --additional-config '{"online_quant_config":{"global_quant_config":"ptpc_fp8","exclude_layer":["lm_head","model.embed_tokens","*self_attn.[qkv]_conv1d*","*block_sparse_moe.experts*","*block_sparse_moe.routed_expert_*","*vision_tower*","*mm_projector*"]}}' 
 ```
 
 The plugin keeps KDA temporal state in fp32, registers every KDA layer through