@@ -60,6 +60,7 @@ def parse_autoep_config(param_dict: dict) -> AutoEPConfig:
     config.route_scale = param_dict.get("route_scale", 1.0)
     config.score_apply = param_dict.get("score_apply", "auto")
     config.combine_impl = param_dict.get("combine_impl", "auto")
+    config.row_weighting_impl = param_dict.get("row_weighting_impl", "auto")
     config.comm_backend = param_dict.get("comm_backend", "comm")
     config.comm_num_sm = param_dict.get("comm_num_sm", 12)
     config.comm_qp_margin = param_dict.get("comm_qp_margin", 4)
@@ -187,12 +188,28 @@ def validate_autoep_config(
         raise ValueError(f"combine_impl must be one of {valid_combine_impl}, "
                          f"got '{config.combine_impl}'")
 
+    # Validate row_weighting_impl
+    valid_row_weighting_impl = ("auto", "eager", "fused")
+    if config.row_weighting_impl not in valid_row_weighting_impl:
+        raise ValueError(f"row_weighting_impl must be one of {valid_row_weighting_impl}, "
+                         f"got '{config.row_weighting_impl}'")
+
     # Validate comm_backend
     valid_comm_backend = ("comm", "deepep")
     if config.comm_backend not in valid_comm_backend:
         raise ValueError(f"comm_backend must be one of {valid_comm_backend}, "
                          f"got '{config.comm_backend}'")
 
+    if config.row_weighting_impl == "fused":
+        if config.comm_backend != "deepep":
+            raise ValueError('row_weighting_impl="fused" only runs inside the DeepEP route, but '
+                             f'comm_backend="{config.comm_backend}" was selected. Set comm_backend to '
+                             '"deepep", or leave row_weighting_impl unset.')
+        if config.autoep_size == 1:
+            raise ValueError('row_weighting_impl="fused" only runs when DeepEP dispatch is active '
+                             '(autoep_size > 1), but autoep_size=1. Increase autoep_size, or leave '
+                             "row_weighting_impl unset.")
+
     # A zero budget would hand the whole GPU to the collective, and a negative
     # one is meaningless; both are worth rejecting where the value is written
     # rather than inside a buffer constructor.
 
@@ -107,6 +107,13 @@ def resolve_combine_impl(
     return "weighted_sum"
 
 
+def resolve_row_weighting_impl(config_override: Literal["auto", "eager", "fused"]) -> Literal["eager", "fused"]:
+    """Resolve DeepEP row weighting implementation from config override."""
+    if config_override != "auto":
+        return config_override
+    return "eager"
+
+
 def _copy_parameter_data(target: nn.Parameter, source: torch.Tensor) -> None:
     full_shape = torch.Size(getattr(source, "ds_shape", source.shape))
     with torch.no_grad():
@@ -158,6 +165,17 @@ def apply_scores_before_experts_if_enabled(
     return routed_input
 
 
+def apply_deepep_row_weights(
+    rows: torch.Tensor,
+    weights: torch.Tensor,
+    row_weighting_impl: Literal["eager", "fused"],
+) -> torch.Tensor:
+    """Apply DeepEP's per-arrival routing weights at the configured boundary."""
+    if row_weighting_impl == "fused":
+        return fused_token_ops.fused_row_weighting(rows, weights)
+    return (rows.float() * weights).to(rows.dtype)
+
+
 def _split_plan_from_expert_counts(
     num_tokens_per_expert: torch.Tensor,  # [E_global], int32
     ep_size: int,
@@ -502,7 +520,9 @@ def __init__(
         self.top_k = spec.top_k
         self.score_apply = resolve_score_apply_mode(spec, config.score_apply)
         self.combine_impl = resolve_combine_impl(config.combine_impl)
+        self.row_weighting_impl = resolve_row_weighting_impl(config.row_weighting_impl)
         self._fused_combine_checked = False
+        self._fused_row_weighting_checked = False
         route_norm = spec.route_norm if config.route_norm is None else config.route_norm
         self.ep_size = ep_size
         self.ep_rank = ep_rank
@@ -819,7 +839,7 @@ def _deepep_route(self, tokens: torch.Tensor, ro: "RouterOutput") -> torch.Tenso
         # doesn't commute with the weight.
         weights = None if recv_weights is None else recv_weights[:arrived].reshape(-1, 1)
         if weights is not None and self.score_apply == "pre":
-            received = (received.float() * weights).to(received.dtype)
+            received = apply_deepep_row_weights(received, weights, self.row_weighting_impl)
             weights = None
 
         # The grouped GEMM needs per-expert row counts, which arrive as a
@@ -834,7 +854,7 @@ def _deepep_route(self, tokens: torch.Tensor, ro: "RouterOutput") -> torch.Tenso
         expert_output = self.experts(received, counts)
 
         if weights is not None:
-            expert_output = (expert_output.float() * weights).to(expert_output.dtype)
+            expert_output = apply_deepep_row_weights(expert_output, weights, self.row_weighting_impl)
 
         return deepep_combine(exchange, expert_output, handle)
 
@@ -909,6 +929,10 @@ def _forward(self, hidden_states: torch.Tensor) -> torch.Tensor | tuple[torch.Te
         if self.combine_impl == "fused_weighted_sum" and not self._fused_combine_checked:
             fused_token_ops.assert_supported(x, score_apply=self.score_apply)
             self._fused_combine_checked = True
+        if self.row_weighting_impl == "fused" and not self._fused_row_weighting_checked:
+            weights = x.new_empty((x.shape[0], 1), dtype=torch.float32)
+            fused_token_ops.assert_row_weighting_supported(x, weights)
+            self._fused_row_weighting_checked = True
 
         # Router
         ro: RouterOutput = RouterOutput(*self.router(x, self.expert_bias))
 
@@ -122,6 +122,7 @@ class AutoEPConfig:
     route_scale: float = 1.0
     score_apply: Literal["auto", "pre", "post"] = "auto"
     combine_impl: Literal["auto", "weighted_sum", "fused_weighted_sum", "legacy_bmm"] = "auto"
+    row_weighting_impl: Literal["auto", "eager", "fused"] = "auto"
     comm_backend: Literal["comm", "deepep"] = "comm"
     comm_num_sm: int = 12
     comm_qp_margin: int = 4
 
@@ -15,6 +15,7 @@
 _IS_ROCM_PYTORCH = getattr(torch.version, "hip", None) is not None
 
 SUPPORTED_ROW_DTYPES = (torch.bfloat16, torch.float16, torch.float32)
+SUPPORTED_ROW_WEIGHTING_DTYPES = (torch.bfloat16, torch.float16)
 
 _MAX_BLOCK_HIDDEN = 512
 # The forward adds one row at a time, so only a single [BLOCK_H] FP32 accumulator is live.
@@ -142,6 +143,84 @@ def _weighted_restore_backward_kernel(
             mask=slot_mask,
         )
 
+    @triton.jit
+    def _row_weighting_forward_kernel(
+        rows_ptr,
+        weights_ptr,
+        out_ptr,
+        hidden,
+        rows_stride,
+        weights_stride,
+        out_stride,
+        BLOCK_H: tl.constexpr,
+    ):
+        row = tl.program_id(0).to(tl.int64)
+        hidden_offsets = tl.program_id(1) * BLOCK_H + tl.arange(0, BLOCK_H)
+        hidden_mask = hidden_offsets < hidden
+
+        values = tl.load(rows_ptr + row * rows_stride + hidden_offsets, mask=hidden_mask, other=0.0).to(tl.float32)
+        weight = tl.load(weights_ptr + row * weights_stride).to(tl.float32)
+        tl.store(
+            out_ptr + row * out_stride + hidden_offsets,
+            (values * weight).to(out_ptr.dtype.element_ty),
+            mask=hidden_mask,
+        )
+
+    @triton.jit
+    def _row_weighting_backward_partial_kernel(
+        grad_out_ptr,
+        rows_ptr,
+        weights_ptr,
+        grad_rows_ptr,
+        partials_ptr,
+        hidden,
+        grad_out_stride,
+        rows_stride,
+        weights_stride,
+        grad_rows_stride,
+        num_hidden_blocks: tl.constexpr,
+        BLOCK_H: tl.constexpr,
+    ):
+        row = tl.program_id(0).to(tl.int64)
+        hidden_block = tl.program_id(1)
+        hidden_offsets = hidden_block * BLOCK_H + tl.arange(0, BLOCK_H)
+        hidden_mask = hidden_offsets < hidden
+
+        upstream = tl.load(
+            grad_out_ptr + row * grad_out_stride + hidden_offsets,
+            mask=hidden_mask,
+            other=0.0,
+        ).to(tl.float32)
+        values = tl.load(
+            rows_ptr + row * rows_stride + hidden_offsets,
+            mask=hidden_mask,
+            other=0.0,
+        ).to(tl.float32)
+        weight = tl.load(weights_ptr + row * weights_stride).to(tl.float32)
+
+        tl.store(
+            grad_rows_ptr + row * grad_rows_stride + hidden_offsets,
+            (upstream * weight).to(grad_rows_ptr.dtype.element_ty),
+            mask=hidden_mask,
+        )
+        partial = tl.sum(upstream * values, axis=0)
+        tl.store(partials_ptr + row * num_hidden_blocks + hidden_block, partial)
+
+    @triton.jit
+    def _row_weighting_backward_reduce_kernel(
+        partials_ptr,
+        grad_weights_ptr,
+        grad_weights_stride,
+        NUM_HIDDEN_BLOCKS: tl.constexpr,
+        BLOCK_B: tl.constexpr,
+    ):
+        row = tl.program_id(0).to(tl.int64)
+        offsets = tl.arange(0, BLOCK_B)
+        mask = offsets < NUM_HIDDEN_BLOCKS
+        partials = tl.load(partials_ptr + row * NUM_HIDDEN_BLOCKS + offsets, mask=mask, other=0.0)
+        grad_weight = tl.sum(partials, axis=0)
+        tl.store(grad_weights_ptr + row * grad_weights_stride, grad_weight)
+
 
 def is_available() -> bool:
     """Whether this build can run the fused weighted restore at all."""
@@ -174,6 +253,16 @@ def _block_hidden(hidden: int, slots: int) -> int:
     return min(_MAX_BLOCK_HIDDEN, budget, max(16, triton.next_power_of_2(hidden)))
 
 
+def _row_weighting_block_hidden(hidden: int) -> int:
+    """Choose a power-of-two hidden tile for row-local products."""
+    return min(_MAX_BLOCK_HIDDEN, max(16, triton.next_power_of_2(hidden)))
+
+
+def _row_weighting_reduce_block(num_hidden_blocks: int) -> int:
+    """Round the per-row partial count up to a power of two for ``tl.arange``."""
+    return max(1, triton.next_power_of_2(num_hidden_blocks))
+
+
 def _padded_top_k(top_k: int) -> int:
     """Round top-k up to a power of two, which ``tl.arange`` requires."""
     return max(2, triton.next_power_of_2(top_k))
@@ -256,6 +345,114 @@ def backward(ctx, grad_output):
         return grad_rows, grad_scores, None, None
 
 
+def assert_row_weighting_supported(rows: torch.Tensor, weights: torch.Tensor) -> None:
+    """Reject unsupported per-row weighting configurations before dispatch."""
+    if not _TRITON_AVAILABLE:
+        raise RuntimeError('row_weighting_impl="fused" needs Triton, which is not installed in this environment. '
+                           'Install Triton, or set row_weighting_impl="eager".')
+    if _IS_ROCM_PYTORCH:
+        raise RuntimeError('row_weighting_impl="fused" is not yet supported on ROCm. Set '
+                           'row_weighting_impl="eager" to run here.')
+    if rows.device.type != "cuda":
+        raise RuntimeError('row_weighting_impl="fused" runs CUDA kernels but rows are on device '
+                           f'"{rows.device.type}". Set row_weighting_impl="eager" to run here.')
+    if weights.device != rows.device:
+        raise RuntimeError('row_weighting_impl="fused" requires rows and weights on the same CUDA device, got '
+                           f"rows on {rows.device} and weights on {weights.device}.")
+    if rows.dim() != 2:
+        raise RuntimeError('row_weighting_impl="fused" requires rows with shape [N, H], got '
+                           f"{tuple(rows.shape)}. Set row_weighting_impl=\"eager\" for unsupported shapes.")
+    expected_weight_shape = (rows.shape[0], 1)
+    if tuple(weights.shape) != expected_weight_shape:
+        raise RuntimeError('row_weighting_impl="fused" requires weights with shape [N, 1] matching rows, got '
+                           f"{tuple(weights.shape)} for rows {tuple(rows.shape)}.")
+    if rows.dtype not in SUPPORTED_ROW_WEIGHTING_DTYPES:
+        raise RuntimeError('row_weighting_impl="fused" supports bfloat16 and float16 rows, got '
+                           f"{rows.dtype}. Set row_weighting_impl=\"eager\", or use a supported dtype.")
+    if weights.dtype != torch.float32:
+        raise RuntimeError('row_weighting_impl="fused" requires float32 weights, got '
+                           f"{weights.dtype}. Set row_weighting_impl=\"eager\" for unsupported weights.")
+    if not rows.is_contiguous():
+        raise RuntimeError('row_weighting_impl="fused" requires contiguous rows. Set row_weighting_impl="eager" '
+                           "for non-contiguous row layouts.")
+    if not weights.is_contiguous():
+        raise RuntimeError('row_weighting_impl="fused" requires contiguous weights. Set row_weighting_impl="eager" '
+                           "for non-contiguous weight layouts.")
+
+
+class _FusedRowWeighting(torch.autograd.Function):
+    """Apply one FP32 routing weight per row, with the eager rounding point."""
+
+    @staticmethod
+    def forward(ctx, rows, weights):
+        output = torch.empty_like(rows)
+        ctx.save_for_backward(rows, weights)
+
+        n_rows, hidden = rows.shape
+        if n_rows == 0 or hidden == 0:
+            return output
+
+        block_hidden = _row_weighting_block_hidden(hidden)
+        grid = (n_rows, triton.cdiv(hidden, block_hidden))
+        _row_weighting_forward_kernel[grid](
+            rows,
+            weights,
+            output,
+            hidden,
+            rows.stride(0),
+            weights.stride(0),
+            output.stride(0),
+            BLOCK_H=block_hidden,
+        )
+        return output
+
+    @staticmethod
+    def backward(ctx, grad_output):
+        rows, weights = ctx.saved_tensors
+        grad_output = grad_output.contiguous()
+        grad_rows = torch.empty_like(rows)
+        grad_weights = torch.empty_like(weights)
+
+        n_rows, hidden = rows.shape
+        if n_rows == 0:
+            return grad_rows, grad_weights
+        if hidden == 0:
+            grad_weights.zero_()
+            return grad_rows, grad_weights
+
+        block_hidden = _row_weighting_block_hidden(hidden)
+        num_hidden_blocks = triton.cdiv(hidden, block_hidden)
+        partials = torch.empty((n_rows, num_hidden_blocks), dtype=torch.float32, device=rows.device)
+        _row_weighting_backward_partial_kernel[(n_rows, num_hidden_blocks)](
+            grad_output,
+            rows,
+            weights,
+            grad_rows,
+            partials,
+            hidden,
+            grad_output.stride(0),
+            rows.stride(0),
+            weights.stride(0),
+            grad_rows.stride(0),
+            num_hidden_blocks=num_hidden_blocks,
+            BLOCK_H=block_hidden,
+        )
+        _row_weighting_backward_reduce_kernel[(n_rows, )](
+            partials,
+            grad_weights,
+            grad_weights.stride(0),
+            NUM_HIDDEN_BLOCKS=num_hidden_blocks,
+            BLOCK_B=_row_weighting_reduce_block(num_hidden_blocks),
+        )
+        return grad_rows, grad_weights
+
+
+def fused_row_weighting(rows: torch.Tensor, weights: torch.Tensor) -> torch.Tensor:
+    """Apply ``weights`` to ``rows`` as ``(rows.float() * weights).to(rows.dtype)``."""
+    assert_row_weighting_supported(rows, weights)
+    return _FusedRowWeighting.apply(rows, weights)
+
+
 def fused_weighted_restore(
     combined_rows: torch.Tensor,
     top_scores: torch.Tensor,
 
@@ -980,6 +980,12 @@ smoke coverage used for this AutoEP surface produced the following version gates
 | -------------------------------------------------------------------------------------------------------------- | -------- |
 | How expert outputs are weighted by their router scores and reduced over top-k. `"auto"` resolves to `"weighted_sum"`. `"fused_weighted_sum"` is experimental and computes the same reduction in one Triton pass, without materializing the scattered assignment buffer or the `[tokens, top_k, hidden]` FP32 intermediate; it requires CUDA, Triton, bfloat16/float16 activations, `tensor_parallel.autotp_size=1`, `expert_tensor_parallel_size=1`, and a resolved `score_apply="post"`, and is rejected rather than silently ignored when any of those does not hold. `"legacy_bmm"` is a debug reduction retained for model-family verification. | `"auto"` |
 
+***row_weighting_impl***: [string]
+
+| Description                                                                                                    | Default  |
+| -------------------------------------------------------------------------------------------------------------- | -------- |
+| How the DeepEP route applies one FP32 routing weight to each received row at the existing `score_apply` boundary. `"auto"` resolves to `"eager"`, preserving `(rows.float() * weights).to(rows.dtype)`. `"fused"` is experimental and uses a separate Triton pointwise operator for that per-row product only; it does not perform the top-k reduction or move the BF16/FP16 rounding point. It requires `comm_backend="deepep"`, `autoep_size > 1`, CUDA, Triton, contiguous bfloat16/float16 rows shaped `[N, H]`, and contiguous FP32 weights shaped `[N, 1]` on the same device; DeepEP dispatch currently supports BF16 rows only. Fused weight gradients can differ from eager due to FP32 summation order. Unsupported configurations fail rather than falling back. | `"auto"` |
+
 ***route_norm***: [boolean]
 
 | Description                                                                                                     | Default |