diff --git a/benchmarks/benchmark_grpo_loss.py b/benchmarks/benchmark_grpo_loss.py
index 027bcb07..6a2c5f01 100644
--- a/benchmarks/benchmark_grpo_loss.py
+++ b/benchmarks/benchmark_grpo_loss.py
@@ -3,14 +3,15 @@
 
 """Benchmark NativeGRPOLossOp vs TritonGRPOLossOp.
 
-Reports forward and forward+backward latency (and peak extra VRAM for the
-forward pass) across a range of (groups x samples x completion-length) shapes.
-The ops operate on per-token log-probs, so the working set scales with
-N = num_prompts * samples_per_prompt * completion_len.
+Both ops consume logits and build on the fused ratio/KL op, so the working set
+scales with the logits tensor [B, T, V] (B = num_prompts * samples_per_prompt).
+Reports forward and forward+backward latency plus peak extra VRAM for the
+forward pass across a range of (groups x samples x completion-length x vocab)
+shapes; the vocab axis is where the online-softmax fusion pays off.
 
 Usage:
     python benchmarks/benchmark_grpo_loss.py
-    python benchmarks/benchmark_grpo_loss.py --iters 50 --clip-eps 0.2 --beta 0.04
+    python benchmarks/benchmark_grpo_loss.py --configs "4,8,256,32768;4,8,256,131072"
 """
 
 import argparse
@@ -19,27 +20,27 @@
 from tabulate import tabulate
 
 from rl_engine.kernels.ops.pytorch.loss.grpo_loss import NativeGRPOLossOp
-from rl_engine.kernels.ops.triton.triton_grpo_loss import TritonGRPOLossOp
+from rl_engine.kernels.ops.triton.loss.grpo_loss import TritonGRPOLossOp
 from rl_engine.platforms.device import device_ctx
 from rl_engine.utils.logger import logger
 
-# (num_prompts, samples_per_prompt, completion_len)
+# (num_prompts, samples_per_prompt, completion_len, vocab)
 DEFAULT_CONFIGS = [
-    (32, 8, 256),
-    (64, 8, 512),
-    (128, 8, 1024),
-    (256, 16, 1024),
+    (4, 8, 256, 32768),
+    (4, 8, 256, 50257),
+    (4, 8, 256, 131072),
 ]
 
 
-def _make_inputs(num_prompts, spp, completion_len, device, dtype):
+def _make_inputs(num_prompts, spp, completion_len, vocab, device, dtype):
     batch = num_prompts * spp
-    current = torch.randn(batch, completion_len, device=device, dtype=dtype)
+    policy = torch.randn(batch, completion_len, vocab, device=device, dtype=dtype)
+    ref = torch.randn(batch, completion_len, vocab, device=device, dtype=dtype)
+    action_ids = torch.randint(0, vocab, (batch, completion_len), device=device)
     old = torch.randn(batch, completion_len, device=device, dtype=dtype)
-    ref = torch.randn(batch, completion_len, device=device, dtype=dtype)
     rewards = torch.randn(batch, device=device, dtype=dtype)
     mask = torch.ones(batch, completion_len, dtype=torch.bool, device=device)
-    return current, old, ref, rewards, mask
+    return policy, ref, action_ids, old, rewards, mask
 
 
 def _time_ms(fn, warmup, iters):
@@ -74,7 +75,7 @@ def run_benchmark(args):
         raise RuntimeError("GRPO loss benchmark requires a CUDA device (Triton op is CUDA-only).")
 
     device = device_ctx.device
-    dtype = torch.float32
+    dtype = torch.float16
     native = NativeGRPOLossOp()
     triton_op = TritonGRPOLossOp()
     kwargs = dict(clip_eps=args.clip_eps, beta=args.beta)
@@ -82,28 +83,30 @@ def run_benchmark(args):
     logger.info(f"GRPO loss benchmark on {device} (dtype={dtype})")
 
     rows = []
-    for num_prompts, spp, comp_len in args.configs:
-        current, old, ref, rewards, mask = _make_inputs(num_prompts, spp, comp_len, device, dtype)
-        n_tokens = current.numel()
-        call_args = (old, ref, rewards, mask)
+    for num_prompts, spp, comp_len, vocab in args.configs:
+        policy, ref, action_ids, old, rewards, mask = _make_inputs(
+            num_prompts, spp, comp_len, vocab, device, dtype
+        )
+        n_tokens = num_prompts * spp * comp_len
+        call_args = (ref, action_ids, old, rewards, mask)
 
-        def native_fwd(c=current):
+        def native_fwd(p=policy):
             with torch.no_grad():
-                native.forward(c, *call_args, samples_per_prompt=spp, **kwargs)
+                native.forward(p, *call_args, samples_per_prompt=spp, **kwargs)
 
-        def triton_fwd(c=current):
+        def triton_fwd(p=policy):
             with torch.no_grad():
-                triton_op.forward(c, *call_args, samples_per_prompt=spp, **kwargs)
+                triton_op.forward(p, *call_args, samples_per_prompt=spp, **kwargs)
 
-        cur_grad = current.clone().requires_grad_(True)
+        pol_grad = policy.clone().requires_grad_(True)
 
-        def native_fwd_bwd(c=cur_grad):
-            loss, _, _ = native.forward(c, *call_args, samples_per_prompt=spp, **kwargs)
-            torch.autograd.grad(loss, c)
+        def native_fwd_bwd(p=pol_grad):
+            loss, _, _ = native.forward(p, *call_args, samples_per_prompt=spp, **kwargs)
+            torch.autograd.grad(loss, p)
 
-        def triton_fwd_bwd(c=cur_grad):
-            loss, _, _ = triton_op.forward(c, *call_args, samples_per_prompt=spp, **kwargs)
-            torch.autograd.grad(loss, c)
+        def triton_fwd_bwd(p=pol_grad):
+            loss, _, _ = triton_op.forward(p, *call_args, samples_per_prompt=spp, **kwargs)
+            torch.autograd.grad(loss, p)
 
         n_fwd = _time_ms(native_fwd, args.warmup, args.iters)
         t_fwd = _time_ms(triton_fwd, args.warmup, args.iters)
@@ -114,21 +117,21 @@ def triton_fwd_bwd(c=cur_grad):
 
         rows.append(
             [
-                f"{num_prompts}x{spp}x{comp_len}",
-                f"{n_tokens/1e6:.2f}M",
+                f"{num_prompts}x{spp}x{comp_len}x{vocab}",
+                f"{n_tokens/1e3:.0f}K",
                 f"{n_fwd:.3f}",
                 f"{t_fwd:.3f}",
                 f"{n_fwd/t_fwd:.2f}x",
                 f"{n_fb:.3f}",
                 f"{t_fb:.3f}",
                 f"{n_fb/t_fb:.2f}x",
-                f"{n_vram*1024:.1f}",
-                f"{t_vram*1024:.1f}",
+                f"{n_vram*1024:.0f}",
+                f"{t_vram*1024:.0f}",
             ]
         )
 
     headers = [
-        "shape (P x S x L)",
+        "shape (P x S x L x V)",
         "tokens",
         "native fwd ms",
         "triton fwd ms",
@@ -144,21 +147,20 @@ def triton_fwd_bwd(c=cur_grad):
 
 def parse_args():
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--iters", type=int, default=30)
-    parser.add_argument("--warmup", type=int, default=10)
+    parser.add_argument("--iters", type=int, default=20)
+    parser.add_argument("--warmup", type=int, default=5)
     parser.add_argument("--clip-eps", type=float, default=0.2)
     parser.add_argument("--beta", type=float, default=0.04)
     parser.add_argument(
         "--configs",
         type=str,
         default=None,
-        help="Semicolon-separated 'prompts,samples,len' triples, e.g. '64,8,512;128,8,1024'.",
+        help="Semicolon-separated 'prompts,samples,len,vocab' tuples, "
+        "e.g. '4,8,256,32768;4,8,256,131072'.",
     )
     args = parser.parse_args()
     if args.configs:
-        args.configs = [
-            tuple(int(x) for x in triple.split(",")) for triple in args.configs.split(";")
-        ]
+        args.configs = [tuple(int(x) for x in tup.split(",")) for tup in args.configs.split(";")]
     else:
         args.configs = DEFAULT_CONFIGS
     return args
diff --git a/benchmarks/benchmark_ratio_kl.py b/benchmarks/benchmark_ratio_kl.py
new file mode 100644
index 00000000..fb172a58
--- /dev/null
+++ b/benchmarks/benchmark_ratio_kl.py
@@ -0,0 +1,340 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026 RL-Kernel Contributors
+
+from __future__ import annotations
+
+import argparse
+import csv
+import statistics
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import torch
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from rl_engine.kernels.ops.pytorch.loss.ratio_kl import NativeRatioKLOp  # noqa: E402
+from rl_engine.testing import make_synthetic_rl_kernel_batch  # noqa: E402
+
+CSV_COLUMNS = [
+    "timestamp",
+    "case",
+    "candidate",
+    "device",
+    "dtype",
+    "num_prompts",
+    "samples_per_prompt",
+    "prompt_len",
+    "completion_len",
+    "vocab_size",
+    "mask_density",
+    "valid_tokens",
+    "reference_ms",
+    "candidate_ms",
+    "speedup",
+    "reference_mem_gb",
+    "candidate_mem_gb",
+    "ratio_drift",
+    "kl_drift",
+    "status",
+    "notes",
+]
+
+
+@dataclass(frozen=True)
+class BenchmarkConfig:
+    case: str
+    candidate: str
+    device: torch.device
+    dtype: torch.dtype
+    num_prompts: int
+    samples_per_prompt: int
+    prompt_len: int
+    completion_len: int
+    vocab_size: int
+    mask_density: float
+    seed: int
+    warmup: int
+    repeat: int
+
+
+def _parse_int_list(value: str) -> list[int]:
+    return [int(item) for item in value.split(",") if item]
+
+
+def _parse_float_list(value: str) -> list[float]:
+    return [float(item) for item in value.split(",") if item]
+
+
+def _parse_dtype(value: str) -> torch.dtype:
+    normalized = value.lower()
+    if normalized in {"fp16", "float16", "half"}:
+        return torch.float16
+    if normalized in {"bf16", "bfloat16"}:
+        return torch.bfloat16
+    if normalized in {"fp32", "float32"}:
+        return torch.float32
+    raise ValueError(f"unsupported dtype: {value}")
+
+
+def _sync(device: torch.device) -> None:
+    if device.type == "cuda":
+        torch.cuda.synchronize(device)
+
+
+def _time_ms(fn, device: torch.device, *, warmup: int = 3, repeat: int = 10) -> tuple[Any, float]:
+    result = None
+    for _ in range(max(0, warmup)):
+        result = fn()
+    _sync(device)
+
+    elapsed: list[float] = []
+    for _ in range(max(1, repeat)):
+        if device.type == "cuda":
+            start = torch.cuda.Event(enable_timing=True)
+            end = torch.cuda.Event(enable_timing=True)
+            start.record()
+            result = fn()
+            end.record()
+            end.synchronize()
+            elapsed.append(start.elapsed_time(end))
+        else:
+            _sync(device)
+            start_time = time.perf_counter()
+            result = fn()
+            _sync(device)
+            elapsed.append((time.perf_counter() - start_time) * 1000.0)
+
+    _sync(device)
+    return result, statistics.median(elapsed)
+
+
+def _peak_memory_gb(device: torch.device) -> float:
+    if device.type != "cuda":
+        return 0.0
+    return torch.cuda.max_memory_allocated(device) / (1024**3)
+
+
+def _reset_peak(device: torch.device) -> None:
+    if device.type == "cuda":
+        torch.cuda.empty_cache()
+        torch.cuda.reset_peak_memory_stats(device)
+
+
+def _ratio_kl_row(config: BenchmarkConfig) -> dict[str, Any]:
+    candidate_name = "TritonRatioKLOp"
+
+    batch = make_synthetic_rl_kernel_batch(
+        num_prompts=config.num_prompts,
+        samples_per_prompt=config.samples_per_prompt,
+        prompt_len=config.prompt_len,
+        completion_len=config.completion_len,
+        vocab_size=config.vocab_size,
+        valid_density=config.mask_density,
+        dtype=config.dtype,
+        device=config.device,
+        seed=config.seed,
+    )
+
+    logit_shape = (batch.batch_size, batch.completion_len, config.vocab_size)
+    policy_logits = torch.randn(logit_shape, device=config.device, dtype=config.dtype)
+    ref_logits = torch.randn(logit_shape, device=config.device, dtype=config.dtype)
+    action_ids = batch.token_ids
+    mask = batch.completion_mask
+    old_logps = batch.old_logps
+
+    native = NativeRatioKLOp()
+
+    def run(op):
+        with torch.no_grad():
+            return op(policy_logits, ref_logits, action_ids, mask, old_logps)
+
+    _reset_peak(config.device)
+    (ref_ratio, ref_kl), reference_ms = _time_ms(
+        lambda: run(native), config.device, warmup=config.warmup, repeat=config.repeat
+    )
+    reference_mem_gb = _peak_memory_gb(config.device)
+
+    notes = ""
+    status = "pass"
+    candidate_ms: float | str = ""
+    speedup: float | str = ""
+    candidate_mem_gb: float | str = ""
+    ratio_drift: float | str = ""
+    kl_drift: float | str = ""
+
+    if config.device.type != "cuda":
+        status = "blocked"
+        notes = "candidate requires CUDA"
+    else:
+        try:
+            from rl_engine.kernels.registry import kernel_registry
+
+            candidate_op = kernel_registry.get_op("ratio_kl")
+            if candidate_op.__class__.__name__ != candidate_name:
+                raise RuntimeError(f"{candidate_name} backend is unavailable")
+
+            _reset_peak(config.device)
+            (cand_ratio, cand_kl), candidate_ms = _time_ms(
+                lambda: run(candidate_op),
+                config.device,
+                warmup=config.warmup,
+                repeat=config.repeat,
+            )
+            candidate_mem_gb = _peak_memory_gb(config.device)
+            speedup = reference_ms / candidate_ms if candidate_ms else float("inf")
+            ratio_drift = (cand_ratio.float() - ref_ratio.float()).abs().max().item()
+            kl_drift = (cand_kl.float() - ref_kl.float()).abs().max().item()
+        except Exception as exc:
+            status = "blocked"
+            notes = f"candidate unavailable: {str(exc).splitlines()[0]}"
+
+    metadata = batch.benchmark_metadata()
+    timing_mode = "cuda_event_median_ms" if config.device.type == "cuda" else "wall_median_ms"
+    timing_notes = f"warmup={config.warmup}; repeat={config.repeat}; {timing_mode}; forward-only"
+    notes = f"{notes}; {timing_notes}" if notes else timing_notes
+
+    def _fmt(value: Any, spec: str) -> Any:
+        return format(value, spec) if isinstance(value, float) else value
+
+    return {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "case": config.case,
+        "candidate": candidate_name,
+        "device": str(config.device),
+        "dtype": str(config.dtype),
+        "num_prompts": config.num_prompts,
+        "samples_per_prompt": config.samples_per_prompt,
+        "prompt_len": config.prompt_len,
+        "completion_len": config.completion_len,
+        "vocab_size": config.vocab_size,
+        "mask_density": config.mask_density,
+        "valid_tokens": metadata["valid_tokens"],
+        "reference_ms": f"{reference_ms:.4f}",
+        "candidate_ms": _fmt(candidate_ms, ".4f"),
+        "speedup": _fmt(speedup, ".2f"),
+        "reference_mem_gb": f"{reference_mem_gb:.6f}",
+        "candidate_mem_gb": _fmt(candidate_mem_gb, ".6f"),
+        "ratio_drift": _fmt(ratio_drift, ".3e"),
+        "kl_drift": _fmt(kl_drift, ".3e"),
+        "status": status,
+        "notes": notes,
+    }
+
+
+def _write_rows(rows: list[dict[str, Any]], output: Path | None) -> None:
+    if output is None:
+        writer = csv.DictWriter(sys.stdout, fieldnames=CSV_COLUMNS)
+        writer.writeheader()
+        writer.writerows(rows)
+        return
+
+    output.parent.mkdir(parents=True, exist_ok=True)
+    exists = output.exists() and output.stat().st_size > 0
+    with output.open("a", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=CSV_COLUMNS)
+        if not exists:
+            writer.writeheader()
+        writer.writerows(rows)
+
+
+def build_arg_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Fused ratio/KL RL-Kernel benchmark runner")
+    parser.add_argument("--case", default="ratio_kl", choices=["ratio_kl"])
+    parser.add_argument("--candidate", default="triton", choices=["triton"])
+    parser.add_argument("--smoke", action="store_true", help="Run a small local-development shape")
+    parser.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu")
+    parser.add_argument("--dtype", default="float16")
+    parser.add_argument("--num-prompts", type=int, default=2)
+    parser.add_argument("--g-sizes", default="4", help="Comma-separated samples-per-prompt values")
+    parser.add_argument("--prompt-len", type=int, default=0)
+    parser.add_argument("--completion-lens", default="512")
+    parser.add_argument("--vocab-sizes", default="32768,131072")
+    parser.add_argument("--mask-densities", default="1.0")
+    parser.add_argument("--seed", type=int, default=0)
+    parser.add_argument("--warmup", type=int, default=3)
+    parser.add_argument("--repeat", type=int, default=10)
+    parser.add_argument("--output", type=Path, default=None)
+    return parser
+
+
+def main() -> None:
+    args = build_arg_parser().parse_args()
+    device = torch.device(args.device)
+    dtype = _parse_dtype(args.dtype)
+
+    if args.smoke:
+        num_prompts = 1
+        g_sizes = [2]
+        prompt_len = 0
+        completion_lens = [8]
+        vocab_sizes = [128]
+        mask_densities = [0.5, 1.0]
+    else:
+        num_prompts = args.num_prompts
+        g_sizes = _parse_int_list(args.g_sizes)
+        prompt_len = args.prompt_len
+        completion_lens = _parse_int_list(args.completion_lens)
+        vocab_sizes = _parse_int_list(args.vocab_sizes)
+        mask_densities = _parse_float_list(args.mask_densities)
+
+    rows: list[dict[str, Any]] = []
+    for samples_per_prompt in g_sizes:
+        for completion_len in completion_lens:
+            for vocab_size in vocab_sizes:
+                for mask_density in mask_densities:
+                    config = BenchmarkConfig(
+                        case=args.case,
+                        candidate=args.candidate,
+                        device=device,
+                        dtype=dtype,
+                        num_prompts=num_prompts,
+                        samples_per_prompt=samples_per_prompt,
+                        prompt_len=prompt_len,
+                        completion_len=completion_len,
+                        vocab_size=vocab_size,
+                        mask_density=mask_density,
+                        seed=args.seed,
+                        warmup=args.warmup,
+                        repeat=args.repeat,
+                    )
+                    try:
+                        rows.append(_ratio_kl_row(config))
+                    except torch.cuda.OutOfMemoryError as exc:
+                        rows.append(
+                            {
+                                "timestamp": datetime.now(timezone.utc).isoformat(),
+                                "case": args.case,
+                                "candidate": "TritonRatioKLOp",
+                                "device": str(device),
+                                "dtype": str(dtype),
+                                "num_prompts": num_prompts,
+                                "samples_per_prompt": samples_per_prompt,
+                                "prompt_len": prompt_len,
+                                "completion_len": completion_len,
+                                "vocab_size": vocab_size,
+                                "mask_density": mask_density,
+                                "valid_tokens": "",
+                                "reference_ms": "",
+                                "candidate_ms": "",
+                                "speedup": "",
+                                "reference_mem_gb": "",
+                                "candidate_mem_gb": "",
+                                "ratio_drift": "",
+                                "kl_drift": "",
+                                "status": "oom",
+                                "notes": str(exc),
+                            }
+                        )
+
+    _write_rows(rows, args.output)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/.nav.yml b/docs/.nav.yml
index 3d9b6291..d9e69b95 100644
--- a/docs/.nav.yml
+++ b/docs/.nav.yml
@@ -10,6 +10,8 @@ nav:
     - Operators:
       - operators/README.md
       - operators/fused-logp.md
+      - operators/grpo-loss.md
+      - operators/ratio-kl.md
       - operators/sampling.md
   - Developer Guide:
     - contributing/README.md
diff --git a/docs/operators/README.md b/docs/operators/README.md
index 49ab34b3..4bb7e9e1 100644
--- a/docs/operators/README.md
+++ b/docs/operators/README.md
@@ -20,5 +20,6 @@ Every operator page should include:
 
 - [Fused LogP](fused-logp.md)
 - [GRPO Loss](grpo-loss.md)
+- [Policy Ratio + KL Penalty](ratio-kl.md)
 - [Sampling](sampling.md)
 - [Operator Doc Template](../contributing/operator-doc-template.md)
diff --git a/docs/operators/grpo-loss.md b/docs/operators/grpo-loss.md
index fb91e8c7..07fcac16 100644
--- a/docs/operators/grpo-loss.md
+++ b/docs/operators/grpo-loss.md
@@ -7,10 +7,12 @@ completion tokens. It targets the GRPO training step, where a naive PyTorch impl
 allocates several broadcasted `[batch, completion_len]` intermediates and a per-token
 advantage tensor.
 
-The operator consumes per-token **log-probs**, so it composes directly with the [Fused LogP](fused-logp.md) operator:
+The operator consumes **logits** directly and builds on the [Policy Ratio + KL
+Penalty](ratio-kl.md) operator: the per-token `policy_ratio` and `kl_penalty` come from the
+fused ratio/KL kernel (logits → ratio/KL via online softmax), and the group-normalized advantages + clipped surrogate are applied on top:
 
 ```
-logits --[logp op]--> logps --[grpo_loss op]--> loss
+logits --[ratio_kl op]--> (ratio, kl) --[group adv + clipped surrogate]--> loss
 ```
 
 ## Entry Point
@@ -20,9 +22,10 @@ from rl_engine.kernels.registry import kernel_registry
 grpo_loss = kernel_registry.get_op("grpo_loss")
 
 loss, policy_loss, kl = grpo_loss(
-    current_logps,        # [B, T] current policy logps (differentiable)
-    old_logps,            # [B, T] inference engine log-probs
-    ref_logps,            # [B, T] reference model log-probs
+    policy_logits,        # [B, T, V] current policy logits (differentiable)
+    ref_logits,           # [B, T, V] frozen reference logits
+    action_ids,           # [B, T] token taken at each position
+    old_logps,            # [B, T] cached behavior-policy log-probs
     rewards,              # [B]
     completion_mask,      # [B, T]
     clip_eps=0.2,
@@ -30,10 +33,11 @@ loss, policy_loss, kl = grpo_loss(
     samples_per_prompt=8,  # uniform groups; or pass group_boundaries=[...]
 )
 
-loss.backward()           # gradient flows into current_logps
+loss.backward()           # gradient flows into policy_logits
 ```
 
-Note: `B = num_prompts * samples_per_prompt`. The [B, T] tensors are made contiguous and flattened to 1-D [N = B*T] before the kernel launch.
+Note: `B = num_prompts * samples_per_prompt`. `old_logps` is the cached behavior-policy
+log-prob from rollout, required for the importance ratio (see [ratio-kl](ratio-kl.md)).
 
 ### Group specification
 
@@ -46,63 +50,70 @@ Provide **exactly one** of:
 
 | Backend | Wrapper | Native symbol | Status |
 | --- | --- | --- | --- |
-| CUDA | `TritonGRPOLossOp` | Triton JIT kernels | Fused forward + analytic backward. |
+| CUDA / ROCm | `TritonGRPOLossOp` | `ratio_kl` + `_group_norm_kernel` | Fused ratio/KL + analytic backward. |
 | PyTorch fallback | `NativeGRPOLossOp` | None | Reference path; CPU and Triton-less GPUs. |
 
-The Triton op fuses three kernels: `_group_norm_kernel` (per-group reward mean/std in
-registers), and token-parallel `_grpo_fwd_kernel` / `_grpo_bwd_kernel`. Each token gathers
-its advantage on the fly (`seq_id = token_index // completion_len`), so the broadcasted
-`[B, T]` advantage tensor is never materialized.
+The Triton op composes the [`ratio_kl`](ratio-kl.md) kernel (per-token `ratio`/`kl` from
+logits, with the analytic backward into `policy_logits`) with the `_group_norm_kernel`
+(per-group reward mean/std in registers). The clipped surrogate + reference-KL reduction is
+a thin autograd-friendly PyTorch layer — no bespoke GRPO loss kernel is needed. The native
+op mirrors this using `NativeRatioKLOp`.
 
 ## Tensor Contract
 
 | Argument | Shape | Dtype | Requirements |
 | --- | --- | --- | --- |
-| `current_logps` | `[B, T]` | float (fp32 recommended) | Differentiable input |
+| `policy_logits` | `[B, T, V]` | float | Differentiable input; contiguous. |
+| `ref_logits` | `[B, T, V]` | float | Constant (no grad); contiguous. |
+| `action_ids` | `[B, T]` | int | Token id per position (in `[0, V)`). |
 | `old_logps` | `[B, T]` | float | Constant (no grad). |
-| `ref_logps` | `[B, T]` | float | Constant (no grad). |
 | `rewards` | `[B]` | float | One scalar per sequence. |
 | `completion_mask` | `[B, T]` | bool / {0,1} | 2-D; `True` marks active tokens. |
 | `loss` (output) | scalar | float32 | `policy_loss + beta * kl`. |
 | `policy_loss`, `kl` (output) | scalar | float32 | Detached reporting values. |
 
+Gradients flow into `policy_logits` only (`ref_logits` is frozen; `old_logps` is cached).
+
 ## Accuracy
 
-Reference semantics (matching `examples/grpo_single_gpu.py`):
+Reference semantics (`NativeGRPOLossOp`):
 
 ```python
 # advantages: group-normalized rewards (population std, unbiased=False)
 grouped = rewards.view(-1, samples_per_prompt)
 adv = (grouped - grouped.mean(1, keepdim=True)) / grouped.std(1, keepdim=True, unbiased=False).clamp_min(1e-6)
-adv = adv.reshape(-1)[:, None].expand_as(completion_mask)
+adv = adv.reshape(-1)[:, None].expand_as(completion_mask).masked_fill(~completion_mask, 0.0)
 
-ratio = torch.exp(current_logps - old_logps)
+# ratio + kl from the ratio_kl op (mask-before-exp; see ratio-kl.md)
+ratio, kl = ratio_kl(policy_logits, ref_logits, action_ids, completion_mask, old_logps)
 policy = -torch.minimum(ratio * adv, torch.clamp(ratio, 1 - clip_eps, 1 + clip_eps) * adv)
-diff = ref_logps - current_logps
-kl = torch.exp(diff) - diff - 1.0                      # k3 estimator
 loss = masked_mean(policy, completion_mask) + beta * masked_mean(kl, completion_mask)
 ```
 
-The Triton op matches the native reference (forward and backward) to `atol=1e-4` in fp32.
-Composing with the dispatched CUDA fused logp matches a torch oracle to `atol=1e-3`.
+The Triton op matches the native reference (forward and backward) to `atol=1e-4`.
 
 ## Performance Notes
 
+The cost is dominated by the [`ratio_kl`](ratio-kl.md) stage (the vocab-dimension work);
+reward normalization and the clipped-surrogate reduction operate on `[B, T]` tensors and are
+negligible.
+
 ```bash
 python benchmarks/benchmark_grpo_loss.py
-python benchmarks/benchmark_grpo_loss.py --configs "64,8,512;256,16,1024"
+python benchmarks/benchmark_grpo_loss.py --configs "4,8,256,32768;4,8,256,131072"
 ```
 
-Indicative results (RTX PRO 6000, SM120, fp32):
+Indicative results (RTX PRO 6000, SM120, fp16, B=32, T=256; native PyTorch vs Triton):
 
-| shape (prompts × samples × len) | tokens | forward speedup | fwd+bwd speedup | peak VRAM (native → Triton) |
-| --- | --- | --- | --- | --- |
-| 64 × 8 × 512 | 0.26M | 4.7× | 3.0× | 10 MB → 1 MB |
-| 128 × 8 × 1024 | 1.05M | 3.2× | 2.7× | 40 MB → 4 MB |
-| 256 × 16 × 1024 | 4.19M | 3.0× | 3.0× | 160 MB → 16 MB |
+| shape (P×S×L×V) | fwd speedup | fwd+bwd speedup | peak fwd VRAM (native → Triton) |
+| --- | --- | --- | --- |
+| 4×8×256×32768 | 5.2× | 2.8× | 2048 MB → ~0 MB |
+| 4×8×256×50257 | 7.3× | 2.4× | 3141 MB → ~0 MB |
+| 4×8×256×131072 | 10.3× | 3.4× | 8192 MB → ~0 MB |
 
-The ~10× VRAM reduction comes from not materializing the broadcasted advantage and
-per-token surrogate/KL intermediates.
+Both speedup and the VRAM advantage grow with vocabulary size: the native path materializes
+the `[B, T, V]` log-softmax (forward peak scales with `V`), while the fused op streams it
+online — the forward peak is independent of `V`.
 
 ## Tests
 
@@ -110,14 +121,15 @@ per-token surrogate/KL intermediates.
 python -m pytest tests/test_grpo_loss.py -v
 ```
 
-Covers the native reference, Triton forward/backward vs native, the
-`logp → grpo_loss` pipeline, masked-token invariance, an SGD loss step, and registry
-dispatch. Triton tests skip without CUDA + Triton.
+Covers the native reference (group advantages + loss from logits), Triton forward/backward
+vs native, masked-token invariance, an SGD loss step, and registry dispatch. Triton tests
+skip without CUDA + Triton.
 
 ## Implementation Files
 
 - `rl_engine/kernels/ops/pytorch/loss/grpo_loss.py`
-- `rl_engine/kernels/ops/triton/triton_grpo_loss.py`
+- `rl_engine/kernels/ops/triton/loss/grpo_loss.py`
+- `rl_engine/kernels/ops/triton/loss/ratio_kl.py`, `rl_engine/kernels/ops/pytorch/loss/ratio_kl.py`
 - `rl_engine/kernels/registry.py`
 - `tests/test_grpo_loss.py`
-- `benchmarks/benchmark_grpo_loss.py`
+- `benchmarks/benchmark_ratio_kl.py`
diff --git a/docs/operators/ratio-kl.md b/docs/operators/ratio-kl.md
new file mode 100644
index 00000000..20629931
--- /dev/null
+++ b/docs/operators/ratio-kl.md
@@ -0,0 +1,107 @@
+# Policy Ratio + KL Penalty
+
+The Ratio/KL operator is the fused front-end of the PPO/GRPO loss: from policy and
+reference **logits** it computes, per active token, the importance `policy_ratio` and the
+reference `kl_penalty` in a single kernel launch.
+
+It is the upstream stage that the GRPO/PPO loss wrappers sit on top of:
+
+```
+logits --[ratio_kl op]--> (ratio, kl) --[clipped surrogate + beta*kl]--> loss
+```
+
+## Entry Point
+```python
+from rl_engine.kernels.registry import kernel_registry
+
+ratio_kl = kernel_registry.get_op("ratio_kl")
+
+policy_ratio, kl_penalty = ratio_kl(
+    policy_logits,    # [B, T, V] current policy logits (differentiable)
+    ref_logits,       # [B, T, V] frozen reference logits
+    action_ids,       # [B, T] token taken at each position
+    attention_mask,   # [B, T] 1 marks active completion tokens
+    old_logps,        # [B, T] cached behavior-policy log-probs
+)
+
+# downstream clipped surrogate (advantages from GAE for PPO / group-norm for GRPO)
+surrogate = -torch.minimum(
+    policy_ratio * advantages,
+    torch.clamp(policy_ratio, 1 - clip_eps, 1 + clip_eps) * advantages,
+)
+loss = masked_mean(surrogate, attention_mask) + beta * masked_mean(kl_penalty, attention_mask)
+loss.backward()       # gradient flows into policy_logits
+```
+
+## Backends
+
+| Backend | Wrapper | Native symbol | Status |
+| --- | --- | --- | --- |
+| CUDA / ROCm | `TritonRatioKLOp` | Triton JIT kernels | Fused forward + analytic backward. |
+| PyTorch fallback | `NativeRatioKLOp` | None | Reference path; CPU and Triton-less GPUs. |
+
+```
+grad_policy_logits[v] = c * (1[v == action] - softmax_policy(v))
+```
+
+so the backward also avoids materializing any `[B, T, V]` probability tensor (only the
+unavoidable `[B, T, V]` gradient output is written).
+
+## Tensor Contract
+
+| Argument | Shape | Dtype | Requirements |
+| --- | --- | --- | --- |
+| `policy_logits` | `[B, T, V]` | float (fp16/bf16/fp32) | Differentiable input; contiguous. |
+| `ref_logits` | `[B, T, V]` | float | Constant (no grad); contiguous. |
+| `action_ids` | `[B, T]` | int | Token id per position (in `[0, V)`). |
+| `attention_mask` | `[B, T]` | bool / {0,1} | `True`/1 marks active tokens. |
+| `old_logps` | `[B, T]` | float | Constant (no grad). |
+| `policy_ratio` (output) | `[B, T]` | float32 | `exp(logp_policy - old_logp)`. |
+| `kl_penalty` (output) | `[B, T]` | float32 | `exp(d) - d - 1`, `d = logp_ref - logp_policy`. |
+
+## Accuracy
+
+Reference semantics (`NativeRatioKLOp`, mask-before-exp matching `grpo_loss`):
+
+```python
+logp_policy = log_softmax(policy_logits, -1).gather(-1, action_ids)   # selected token logp
+with torch.no_grad():
+    logp_ref = log_softmax(ref_logits, -1).gather(-1, action_ids)
+
+delta = (logp_policy - old_logps).masked_fill(~mask, 0.0)
+diff  = (logp_ref - logp_policy).masked_fill(~mask, 0.0)
+policy_ratio = torch.exp(delta)                      # exp(0) = 1 on inactive tokens
+kl_penalty   = torch.exp(diff) - diff - 1.0          # k3 estimator
+```
+
+The Triton op matches the native reference on `ratio` and `kl` (forward) and on the
+`policy_logits` gradient (backward) to `atol=1e-4` (fp32). In fp16 the KL difference is
+~`1e-4` from rounding; the ratio difference is ~`1e-9`.
+
+## Performance Notes
+
+```bash
+python benchmarks/benchmark_ratio_kl.py
+python benchmarks/benchmark_ratio_kl.py --g-sizes 8 --completion-lens 512 --vocab-sizes 32768,131072
+```
+
+Indicative forward-only results (fp16, `B=16`, `T=512`):
+
+| vocab | active tokens | forward speedup | peak VRAM (native → Triton) |
+| --- | --- | --- | --- |
+| 32,768 | 8,192 | 6.8× | 3.0 GB → 1.0 GB |
+| 131,072 | 8,192 | 10.0× | 12.0 GB → 4.0 GB |
+
+## Tests
+
+```bash
+python -m pytest tests/test_ratio_kl.py -v
+```
+
+## Implementation Files
+
+- `rl_engine/kernels/ops/pytorch/loss/ratio_kl.py`
+- `rl_engine/kernels/ops/triton/loss/ratio_kl.py`
+- `rl_engine/kernels/registry.py`
+- `tests/test_ratio_kl.py`
+- `benchmarks/benchmark_ratio_kl.py`
diff --git a/rl_engine/kernels/ops/pytorch/loss/grpo_loss.py b/rl_engine/kernels/ops/pytorch/loss/grpo_loss.py
index 8161666a..d0f11bfc 100644
--- a/rl_engine/kernels/ops/pytorch/loss/grpo_loss.py
+++ b/rl_engine/kernels/ops/pytorch/loss/grpo_loss.py
@@ -7,18 +7,26 @@
 
 import torch
 
+from rl_engine.kernels.ops.pytorch.loss.ratio_kl import NativeRatioKLOp
+
 
 class NativeGRPOLossOp:
-    """Pure PyTorch native fallback for the fused GRPO loss."""
+    """Pure PyTorch native fallback for the fused GRPO loss.
+
+    Consumes logits directly: the per-token ``policy_ratio`` / ``kl_penalty`` come
+    from the fused ratio/KL op, and the group-normalized advantages + clipped
+    surrogate + reference-KL reduction are applied on top.
+    """
 
     def __init__(self) -> None:
-        pass
+        self._ratio_kl = NativeRatioKLOp()
 
     def __call__(
         self,
-        current_logps: torch.Tensor,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
         old_logps: torch.Tensor,
-        ref_logps: torch.Tensor,
         rewards: torch.Tensor,
         completion_mask: torch.Tensor,
         *,
@@ -29,9 +37,10 @@ def __call__(
         eps: float = 1e-6,
     ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
         return self.forward(
-            current_logps,
+            policy_logits,
+            ref_logits,
+            action_ids,
             old_logps,
-            ref_logps,
             rewards,
             completion_mask,
             clip_eps=clip_eps,
@@ -86,38 +95,41 @@ def expand_advantages(
 
     def apply(
         self,
-        current_logps: torch.Tensor,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
         old_logps: torch.Tensor,
-        ref_logps: torch.Tensor,
-        advantages: torch.Tensor,
+        sample_advantages: torch.Tensor,
         completion_mask: torch.Tensor,
         *,
         clip_eps: float = 0.2,
         beta: float = 0.0,
     ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
-        """Compute ``(loss, policy_loss, kl)`` from per-token log-probabilities."""
+        """Compute ``(loss, policy_loss, kl)`` from logits + per-sequence advantages.
+
+        ``sample_advantages`` is per-sequence and is broadcast to per-token here.
+        The ratio and KL come from the fused ratio/KL op; gradients flow into
+        ``policy_logits``.
+        """
+        ratio, kl_terms = self._ratio_kl(
+            policy_logits, ref_logits, action_ids, completion_mask, old_logps
+        )
         bool_mask = completion_mask.bool()
-
-        delta = (current_logps.float() - old_logps.float()).masked_fill(~bool_mask, 0.0)
-        diff = (ref_logps.float() - current_logps.float()).masked_fill(~bool_mask, 0.0)
-        ratio = torch.exp(delta)
-        adv = advantages.float()
+        adv = self.expand_advantages(sample_advantages, completion_mask).float()
         unclipped = ratio * adv
         clipped = torch.clamp(ratio, 1.0 - clip_eps, 1.0 + clip_eps) * adv
         policy_loss_terms = -torch.minimum(unclipped, clipped)
 
-        # k3 reference-KL estimator: exp(d) - d - 1, with d = ref - current.
-        kl_terms = torch.exp(diff) - diff - 1.0
-
         policy_loss = self._masked_mean(policy_loss_terms, bool_mask)
         kl = self._masked_mean(kl_terms, bool_mask)
         return policy_loss + beta * kl, policy_loss, kl
 
     def forward(
         self,
-        current_logps: torch.Tensor,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
         old_logps: torch.Tensor,
-        ref_logps: torch.Tensor,
         rewards: torch.Tensor,
         completion_mask: torch.Tensor,
         *,
@@ -133,12 +145,12 @@ def forward(
             group_boundaries=group_boundaries,
             eps=eps,
         )
-        advantages = self.expand_advantages(sample_advantages, completion_mask)
         return self.apply(
-            current_logps,
+            policy_logits,
+            ref_logits,
+            action_ids,
             old_logps,
-            ref_logps,
-            advantages,
+            sample_advantages,
             completion_mask,
             clip_eps=clip_eps,
             beta=beta,
diff --git a/rl_engine/kernels/ops/pytorch/loss/ratio_kl.py b/rl_engine/kernels/ops/pytorch/loss/ratio_kl.py
new file mode 100644
index 00000000..1d7f0126
--- /dev/null
+++ b/rl_engine/kernels/ops/pytorch/loss/ratio_kl.py
@@ -0,0 +1,56 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026 RL-Kernel Contributors
+
+from __future__ import annotations
+
+from typing import Tuple
+
+import torch
+
+from rl_engine.kernels.ops.pytorch.loss.logp import NativeLogpOp
+
+
+class NativeRatioKLOp:
+    """PyTorch native fallback for the fused ratio + KL operator."""
+
+    def __init__(self) -> None:
+        self._logp = NativeLogpOp()
+
+    def __call__(
+        self,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
+        attention_mask: torch.Tensor,
+        old_logps: torch.Tensor,
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        return self.forward(policy_logits, ref_logits, action_ids, attention_mask, old_logps)
+
+    def _selected_logp(self, logits: torch.Tensor, action_ids: torch.Tensor) -> torch.Tensor:
+        return self._logp.apply_fp32(logits, action_ids)
+
+    def forward(
+        self,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
+        attention_mask: torch.Tensor,
+        old_logps: torch.Tensor,
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        mask = attention_mask.to(torch.bool)
+        vocab_size = policy_logits.size(-1)
+        invalid = mask & ((action_ids < 0) | (action_ids >= vocab_size))
+        if invalid.any():
+            raise ValueError(
+                f"action_ids at active (unmasked) positions must be in "
+                f"[0, {vocab_size}); found out-of-range ids."
+            )
+        logp_policy = self._selected_logp(policy_logits, action_ids)
+        with torch.no_grad():
+            logp_ref = self._selected_logp(ref_logits, action_ids)
+
+        delta = (logp_policy - old_logps.float()).masked_fill(~mask, 0.0)
+        diff = (logp_ref - logp_policy).masked_fill(~mask, 0.0)
+        ratio = torch.exp(delta)
+        kl = torch.exp(diff) - diff - 1.0
+        return ratio, kl
diff --git a/rl_engine/kernels/ops/triton/loss/__init__.py b/rl_engine/kernels/ops/triton/loss/__init__.py
new file mode 100644
index 00000000..86cf4c9d
--- /dev/null
+++ b/rl_engine/kernels/ops/triton/loss/__init__.py
@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026 RL-Kernel Contributors
diff --git a/rl_engine/kernels/ops/triton/triton_grpo_loss.py b/rl_engine/kernels/ops/triton/loss/grpo_loss.py
similarity index 51%
rename from rl_engine/kernels/ops/triton/triton_grpo_loss.py
rename to rl_engine/kernels/ops/triton/loss/grpo_loss.py
index 8beadc74..d6b402d8 100644
--- a/rl_engine/kernels/ops/triton/triton_grpo_loss.py
+++ b/rl_engine/kernels/ops/triton/loss/grpo_loss.py
@@ -8,7 +8,7 @@
 import triton
 import triton.language as tl
 
-_BLOCK = 1024
+from rl_engine.kernels.ops.triton.loss.ratio_kl import TritonRatioKLOp
 
 
 def _next_pow2(x: int) -> int:
@@ -55,182 +55,26 @@ def _group_norm_kernel(
     tl.store(adv_ptr + start + offs, adv, mask=keep)
 
 
-@triton.jit
-def _grpo_fwd_kernel(
-    cur_ptr,
-    old_ptr,
-    ref_ptr,
-    adv_seq_ptr,  # float32[B], per-sequence advantages
-    mask_ptr,
-    partials_ptr,  # float32[grid, 2]: per-block (policy_sum, kl_sum)
-    n_elements,
-    T,  # completion length (tokens per sequence)
-    clip_eps,
-    BLOCK: tl.constexpr,
-):
-    pid = tl.program_id(0)
-    offs = pid * BLOCK + tl.arange(0, BLOCK)
-    bound = offs < n_elements
-    seq_id = offs // T
-
-    cur = tl.load(cur_ptr + offs, mask=bound, other=0.0).to(tl.float32)
-    old = tl.load(old_ptr + offs, mask=bound, other=0.0).to(tl.float32)
-    ref = tl.load(ref_ptr + offs, mask=bound, other=0.0).to(tl.float32)
-    adv = tl.load(adv_seq_ptr + seq_id, mask=bound, other=0.0).to(tl.float32)
-    active = tl.load(mask_ptr + offs, mask=bound, other=0).to(tl.float32)
-    keep = bound & (active != 0.0)
-
-    ratio = tl.exp(cur - old)
-    lo = 1.0 - clip_eps
-    hi = 1.0 + clip_eps
-    unclipped = ratio * adv
-    clipped = tl.minimum(tl.maximum(ratio, lo), hi) * adv
-    policy_term = -tl.minimum(unclipped, clipped)
-
-    diff = ref - cur
-    kl_term = tl.exp(diff) - diff - 1.0
-
-    policy_term = tl.where(keep, policy_term, 0.0)
-    kl_term = tl.where(keep, kl_term, 0.0)
-
-    tl.store(partials_ptr + pid * 2 + 0, tl.sum(policy_term, axis=0))
-    tl.store(partials_ptr + pid * 2 + 1, tl.sum(kl_term, axis=0))
-
-
-@triton.jit
-def _grpo_bwd_kernel(
-    cur_ptr,
-    old_ptr,
-    ref_ptr,
-    adv_seq_ptr,
-    mask_ptr,
-    grad_cur_ptr,
-    scale,  # grad_output / num_active
-    beta,
-    clip_eps,
-    n_elements,
-    T,
-    BLOCK: tl.constexpr,
-):
-    pid = tl.program_id(0)
-    offs = pid * BLOCK + tl.arange(0, BLOCK)
-    bound = offs < n_elements
-    seq_id = offs // T
-
-    cur = tl.load(cur_ptr + offs, mask=bound, other=0.0).to(tl.float32)
-    old = tl.load(old_ptr + offs, mask=bound, other=0.0).to(tl.float32)
-    ref = tl.load(ref_ptr + offs, mask=bound, other=0.0).to(tl.float32)
-    adv = tl.load(adv_seq_ptr + seq_id, mask=bound, other=0.0).to(tl.float32)
-    active = tl.load(mask_ptr + offs, mask=bound, other=0).to(tl.float32)
-    keep = bound & (active != 0.0)
-
-    ratio = tl.exp(cur - old)
-    lo = 1.0 - clip_eps
-    hi = 1.0 + clip_eps
-    unclipped = ratio * adv
-    clipped = tl.minimum(tl.maximum(ratio, lo), hi) * adv
-
-    # d(ratio)/d(cur) = ratio. The surrogate selects the smaller branch; the
-    # clamped branch has zero gradient outside (lo, hi).
-    in_range = (ratio > lo) & (ratio < hi)
-    d_clipped = tl.where(in_range, ratio * adv, 0.0)
-    sel_unclipped = unclipped <= clipped
-    deriv_sel = tl.where(sel_unclipped, ratio * adv, d_clipped)
-    d_policy = -deriv_sel
-
-    # d(kl)/d(cur) for kl = exp(ref - cur) - (ref - cur) - 1.
-    d_kl = 1.0 - tl.exp(ref - cur)
-
-    grad = scale * (d_policy + beta * d_kl)
-    grad = tl.where(keep, grad, 0.0)
-    tl.store(grad_cur_ptr + offs, grad, mask=bound)
-
-
-class _GRPOLossFunction(torch.autograd.Function):
-    """Autograd wrapper around the token-parallel Triton kernels."""
-
-    @staticmethod
-    def forward(
-        ctx, current_logps, old_logps, ref_logps, adv_seq, mask, completion_len, clip_eps, beta
-    ):
-        cur = current_logps.contiguous()
-        old = old_logps.contiguous().to(cur.dtype)
-        ref = ref_logps.contiguous().to(cur.dtype)
-        adv = adv_seq.contiguous().to(torch.float32)
-        mask_f = mask.contiguous().to(torch.float32)
-
-        n = cur.numel()
-        num_active = mask_f.sum().clamp_min(1e-8)
-
-        grid = (triton.cdiv(n, _BLOCK),)
-        partials = torch.empty(grid[0], 2, device=cur.device, dtype=torch.float32)
-        _grpo_fwd_kernel[grid](
-            cur.reshape(-1),
-            old.reshape(-1),
-            ref.reshape(-1),
-            adv,
-            mask_f.reshape(-1),
-            partials,
-            n,
-            int(completion_len),
-            float(clip_eps),
-            BLOCK=_BLOCK,
-        )
-
-        block_sums = partials.sum(dim=0)
-        policy_loss = block_sums[0] / num_active
-        kl = block_sums[1] / num_active
-        loss = policy_loss + beta * kl
-
-        ctx.save_for_backward(cur, old, ref, adv, mask_f)
-        ctx.clip_eps = float(clip_eps)
-        ctx.beta = float(beta)
-        ctx.completion_len = int(completion_len)
-        ctx.num_active = num_active
-        ctx.mark_non_differentiable(policy_loss, kl)
-        return loss, policy_loss, kl
-
-    @staticmethod
-    def backward(ctx, grad_loss, grad_policy, grad_kl):
-        cur, old, ref, adv, mask_f = ctx.saved_tensors
-        n = cur.numel()
-        grad_cur = torch.empty_like(cur, dtype=torch.float32)
-        scale = float((grad_loss / ctx.num_active).item())
-
-        grid = (triton.cdiv(n, _BLOCK),)
-        _grpo_bwd_kernel[grid](
-            cur.reshape(-1),
-            old.reshape(-1),
-            ref.reshape(-1),
-            adv,
-            mask_f.reshape(-1),
-            grad_cur.reshape(-1),
-            scale,
-            ctx.beta,
-            ctx.clip_eps,
-            n,
-            ctx.completion_len,
-            BLOCK=_BLOCK,
-        )
-
-        grad_cur = grad_cur.to(cur.dtype)
-        # Inputs: current, old, ref, adv_seq, mask, completion_len, clip_eps, beta.
-        return grad_cur, None, None, None, None, None, None, None
-
-
 class TritonGRPOLossOp:
     """Triton fused GRPO loss op.
 
-    ``forward`` is the drop-in equivalent of ``NativeGRPOLossOp.forward`` (raw
-    rewards in, scalar loss out). ``apply`` takes the per-sequence advantage
-    vector directly (the fused representation), not a per-token tensor.
+    The per-token ``policy_ratio`` / ``kl_penalty`` are produced by the fused
+    ``ratio_kl`` Triton kernel (logits -> ratio/KL via online softmax, with an
+    analytic backward into ``policy_logits``); reward normalization runs in the
+    ``_group_norm_kernel``; the clipped surrogate + reference-KL reduction are a
+    thin autograd-friendly PyTorch layer on top. ``forward`` takes raw rewards;
+    ``apply`` takes the per-sequence advantage vector directly.
     """
 
+    def __init__(self) -> None:
+        self._ratio_kl = TritonRatioKLOp()
+
     def __call__(
         self,
-        current_logps: torch.Tensor,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
         old_logps: torch.Tensor,
-        ref_logps: torch.Tensor,
         rewards: torch.Tensor,
         completion_mask: torch.Tensor,
         *,
@@ -241,9 +85,10 @@ def __call__(
         eps: float = 1e-6,
     ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
         return self.forward(
-            current_logps,
+            policy_logits,
+            ref_logits,
+            action_ids,
             old_logps,
-            ref_logps,
             rewards,
             completion_mask,
             clip_eps=clip_eps,
@@ -319,39 +164,61 @@ def group_advantages(
         )
         return adv
 
+    @staticmethod
+    def expand_advantages(
+        sample_advantages: torch.Tensor,
+        completion_mask: torch.Tensor,
+    ) -> torch.Tensor:
+        """Broadcast per-sequence advantages to per-token, zeroing masked tokens."""
+        bool_mask = completion_mask.bool()
+        expanded = sample_advantages.reshape(-1, 1).expand_as(bool_mask).clone()
+        return expanded.masked_fill(~bool_mask, 0.0)
+
+    @staticmethod
+    def _masked_mean(
+        values: torch.Tensor, bool_mask: torch.Tensor, eps: float = 1e-8
+    ) -> torch.Tensor:
+        masked = values.masked_fill(~bool_mask, 0.0)
+        denom = bool_mask.sum().to(values.dtype).clamp_min(eps)
+        return masked.sum() / denom
+
     def apply(
         self,
-        current_logps: torch.Tensor,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
         old_logps: torch.Tensor,
-        ref_logps: torch.Tensor,
         sample_advantages: torch.Tensor,
         completion_mask: torch.Tensor,
         *,
         clip_eps: float = 0.2,
         beta: float = 0.0,
     ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
-        """Evaluate the loss from per-sequence advantages (gathered per token)."""
-        if not current_logps.is_cuda:
+        """Evaluate the loss from logits + per-sequence advantages."""
+        if not policy_logits.is_cuda:
             raise RuntimeError("TritonGRPOLossOp requires CUDA tensors.")
         if completion_mask.ndim != 2:
             raise ValueError("completion_mask must be 2D [num_sequences, completion_len].")
-        completion_len = completion_mask.shape[1]
-        return _GRPOLossFunction.apply(
-            current_logps,
-            old_logps,
-            ref_logps,
-            sample_advantages,
-            completion_mask,
-            completion_len,
-            clip_eps,
-            beta,
+
+        ratio, kl_terms = self._ratio_kl(
+            policy_logits, ref_logits, action_ids, completion_mask, old_logps
         )
+        bool_mask = completion_mask.bool()
+        adv = self.expand_advantages(sample_advantages, completion_mask).float()
+        unclipped = ratio * adv
+        clipped = torch.clamp(ratio, 1.0 - clip_eps, 1.0 + clip_eps) * adv
+        policy_loss_terms = -torch.minimum(unclipped, clipped)
+
+        policy_loss = self._masked_mean(policy_loss_terms, bool_mask)
+        kl = self._masked_mean(kl_terms, bool_mask)
+        return policy_loss + beta * kl, policy_loss, kl
 
     def forward(
         self,
-        current_logps: torch.Tensor,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
         old_logps: torch.Tensor,
-        ref_logps: torch.Tensor,
         rewards: torch.Tensor,
         completion_mask: torch.Tensor,
         *,
@@ -368,9 +235,10 @@ def forward(
             eps=eps,
         )
         return self.apply(
-            current_logps,
+            policy_logits,
+            ref_logits,
+            action_ids,
             old_logps,
-            ref_logps,
             sample_advantages,
             completion_mask,
             clip_eps=clip_eps,
diff --git a/rl_engine/kernels/ops/triton/loss/ratio_kl.py b/rl_engine/kernels/ops/triton/loss/ratio_kl.py
new file mode 100644
index 00000000..59ac5ca0
--- /dev/null
+++ b/rl_engine/kernels/ops/triton/loss/ratio_kl.py
@@ -0,0 +1,196 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026 RL-Kernel Contributors
+
+"""Fused Policy-ratio + KL-penalty Triton operator (PPO/GRPO front-end).
+policy_ratio = exp(logp_policy - old_logp)
+kl_penalty   = exp(d) - d - 1,   d = logp_ref - logp_policy   (k3 estimator)
+"""
+
+from __future__ import annotations
+
+from typing import Tuple
+
+import torch
+import triton
+import triton.language as tl
+
+# Tile width
+_MAX_BLOCK_V = 2048
+
+
+@triton.jit
+def _ratio_kl_fwd_kernel(
+    policy_ptr,
+    ref_ptr,
+    action_ptr,
+    mask_ptr,
+    old_ptr,
+    ratio_ptr,
+    kl_ptr,
+    diff_ptr,  # saved for backward: d = logp_ref - logp_policy
+    logz_ptr,  # saved for backward: log-sum-exp of policy logits
+    V,
+    BLOCK_V: tl.constexpr,
+):
+    row = tl.program_id(0)
+    active = tl.load(mask_ptr + row) != 0
+
+    if active:
+        row_off = row.to(tl.int64) * V
+        a = tl.load(action_ptr + row)
+
+        max_p = -float("inf")
+        sum_p = 0.0
+        max_r = -float("inf")
+        sum_r = 0.0
+        for start in range(0, V, BLOCK_V):
+            cols = start + tl.arange(0, BLOCK_V)
+            cmask = cols < V
+            p = tl.load(policy_ptr + row_off + cols, mask=cmask, other=-float("inf")).to(tl.float32)
+            r = tl.load(ref_ptr + row_off + cols, mask=cmask, other=-float("inf")).to(tl.float32)
+
+            tmax_p = tl.max(p, axis=0)
+            nmax_p = tl.maximum(max_p, tmax_p)
+            sum_p = sum_p * tl.exp(max_p - nmax_p) + tl.sum(tl.exp(p - nmax_p), axis=0)
+            max_p = nmax_p
+
+            tmax_r = tl.max(r, axis=0)
+            nmax_r = tl.maximum(max_r, tmax_r)
+            sum_r = sum_r * tl.exp(max_r - nmax_r) + tl.sum(tl.exp(r - nmax_r), axis=0)
+            max_r = nmax_r
+
+        logz_p = max_p + tl.log(sum_p)
+        logz_r = max_r + tl.log(sum_r)
+        pa = tl.load(policy_ptr + row_off + a).to(tl.float32)
+        ra = tl.load(ref_ptr + row_off + a).to(tl.float32)
+        logp_p = pa - logz_p
+        logp_r = ra - logz_r
+
+        old = tl.load(old_ptr + row).to(tl.float32)
+        ratio = tl.exp(logp_p - old)
+        d = logp_r - logp_p
+        kl = tl.exp(d) - d - 1.0
+
+        tl.store(ratio_ptr + row, ratio)
+        tl.store(kl_ptr + row, kl)
+        tl.store(diff_ptr + row, d)
+        tl.store(logz_ptr + row, logz_p)
+    else:
+        # Inactive token
+        tl.store(ratio_ptr + row, 1.0)
+        tl.store(kl_ptr + row, 0.0)
+        tl.store(diff_ptr + row, 0.0)
+        tl.store(logz_ptr + row, 0.0)
+
+
+@triton.jit
+def _ratio_kl_bwd_kernel(
+    policy_ptr,
+    action_ptr,
+    mask_ptr,
+    ratio_ptr,
+    diff_ptr,
+    logz_ptr,
+    grad_ratio_ptr,
+    grad_kl_ptr,
+    grad_policy_ptr,  # [N, V] fp32, pre-zeroed
+    V,
+    BLOCK_V: tl.constexpr,
+):
+    row = tl.program_id(0)
+    active = tl.load(mask_ptr + row) != 0
+    if active:
+        row_off = row.to(tl.int64) * V
+        a = tl.load(action_ptr + row)
+        ratio = tl.load(ratio_ptr + row)
+        d = tl.load(diff_ptr + row)
+        logz = tl.load(logz_ptr + row)
+        g_ratio = tl.load(grad_ratio_ptr + row)
+        g_kl = tl.load(grad_kl_ptr + row)
+
+        # d(ratio)/d(logp_p) = ratio ; d(kl)/d(logp_p) = 1 - exp(d).
+        # Both chain through (1[v==a] - softmax_policy(v)).
+        c = g_ratio * ratio + g_kl * (1.0 - tl.exp(d))
+
+        for start in range(0, V, BLOCK_V):
+            cols = start + tl.arange(0, BLOCK_V)
+            cmask = cols < V
+            p = tl.load(policy_ptr + row_off + cols, mask=cmask, other=0.0).to(tl.float32)
+            soft = tl.exp(p - logz)
+            onehot = tl.where(cols == a, 1.0, 0.0)
+            grad = c * (onehot - soft)
+            tl.store(grad_policy_ptr + row_off + cols, grad, mask=cmask)
+
+
+class _RatioKLFunction(torch.autograd.Function):
+    @staticmethod
+    def forward(ctx, policy_logits, ref_logits, action_ids, attention_mask, old_logps):
+        if not policy_logits.is_cuda:
+            raise RuntimeError("TritonRatioKLOp requires CUDA/ROCm tensors.")
+        expected = tuple(policy_logits.shape[:-1])
+        for name, t in (
+            ("action_ids", action_ids),
+            ("attention_mask", attention_mask),
+            ("old_logps", old_logps),
+        ):
+            if tuple(t.shape) != expected:
+                raise ValueError(
+                    f"{name} shape {tuple(t.shape)} does not match "
+                    f"policy_logits.shape[:-1] {expected}."
+                )
+        V = policy_logits.shape[-1]
+        pol = policy_logits.contiguous().view(-1, V)
+        ref = ref_logits.contiguous().view(-1, V)
+        n_rows = pol.shape[0]
+        act = action_ids.contiguous().view(-1).clamp(0, V - 1).to(torch.int64)
+        mask = attention_mask.contiguous().view(-1).to(torch.int32)
+        old = old_logps.contiguous().view(-1).to(torch.float32)
+
+        ratio = torch.empty(n_rows, device=pol.device, dtype=torch.float32)
+        kl = torch.empty(n_rows, device=pol.device, dtype=torch.float32)
+        diff = torch.empty(n_rows, device=pol.device, dtype=torch.float32)
+        logz = torch.empty(n_rows, device=pol.device, dtype=torch.float32)
+
+        block_v = min(_MAX_BLOCK_V, triton.next_power_of_2(V))
+        _ratio_kl_fwd_kernel[(n_rows,)](
+            pol, ref, act, mask, old, ratio, kl, diff, logz, V, BLOCK_V=block_v
+        )
+
+        ctx.save_for_backward(pol, act, mask, ratio, diff, logz)
+        ctx.block_v = block_v
+        ctx.policy_shape = tuple(policy_logits.shape)
+        ctx.policy_dtype = policy_logits.dtype
+        ctx.out_shape = tuple(attention_mask.shape)
+        return ratio.view(ctx.out_shape), kl.view(ctx.out_shape)
+
+    @staticmethod
+    def backward(ctx, grad_ratio, grad_kl):
+        pol, act, mask, ratio, diff, logz = ctx.saved_tensors
+        n_rows, V = pol.shape
+        gr = grad_ratio.contiguous().view(-1).to(torch.float32)
+        gk = grad_kl.contiguous().view(-1).to(torch.float32)
+        grad_pol = torch.zeros_like(pol, dtype=torch.float32)
+
+        _ratio_kl_bwd_kernel[(n_rows,)](
+            pol, act, mask, ratio, diff, logz, gr, gk, grad_pol, V, BLOCK_V=ctx.block_v
+        )
+
+        grad_pol = grad_pol.view(ctx.policy_shape).to(ctx.policy_dtype)
+        # policy_logits, ref_logits, action_ids, attention_mask, old_logps
+        return grad_pol, None, None, None, None
+
+
+class TritonRatioKLOp:
+    """Fused policy-ratio + KL-penalty op (Triton; CUDA & ROCm)."""
+
+    def __call__(
+        self,
+        policy_logits: torch.Tensor,
+        ref_logits: torch.Tensor,
+        action_ids: torch.Tensor,
+        attention_mask: torch.Tensor,
+        old_logps: torch.Tensor,
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        return _RatioKLFunction.apply(
+            policy_logits, ref_logits, action_ids, attention_mask, old_logps
+        )
diff --git a/rl_engine/kernels/registry.py b/rl_engine/kernels/registry.py
index 7d85d263..7aae08fb 100644
--- a/rl_engine/kernels/registry.py
+++ b/rl_engine/kernels/registry.py
@@ -34,9 +34,13 @@ class OpBackend(Enum, metaclass=_KernelEnumMeta):
     ROCM_CK = "rl_engine.kernels.ops.rocm.composable_kernel.CKOp"
 
     # GRPO loss (group reward normalization + clipped surrogate + KL)
-    TRITON_GRPO_LOSS = "rl_engine.kernels.ops.triton.triton_grpo_loss.TritonGRPOLossOp"
+    TRITON_GRPO_LOSS = "rl_engine.kernels.ops.triton.loss.grpo_loss.TritonGRPOLossOp"
     PYTORCH_GRPO_LOSS = "rl_engine.kernels.ops.pytorch.loss.grpo_loss.NativeGRPOLossOp"
 
+    # Fused policy-ratio + KL-penalty front-end (PPO/GRPO), logits -> (ratio, kl)
+    TRITON_RATIO_KL = "rl_engine.kernels.ops.triton.loss.ratio_kl.TritonRatioKLOp"
+    PYTORCH_RATIO_KL = "rl_engine.kernels.ops.pytorch.loss.ratio_kl.NativeRatioKLOp"
+
     # Generic fallback
     TRITON_GENERIC = "rl_engine.kernels.ops.triton.generic.TritonOp"
     PYTORCH_NATIVE = "rl_engine.kernels.ops.pytorch.loss.logp.NativeLogpOp"
@@ -74,17 +78,20 @@ def __init__(self):
                 ],
                 "attn": [OpBackend.FLASH_ATTN, OpBackend.TRITON_GENERIC, OpBackend.PYTORCH_NATIVE],
                 "grpo_loss": [OpBackend.TRITON_GRPO_LOSS, OpBackend.PYTORCH_GRPO_LOSS],
+                "ratio_kl": [OpBackend.TRITON_RATIO_KL, OpBackend.PYTORCH_RATIO_KL],
                 # Default dispatch logic for new operators
             },
             "rocm": {
                 "logp": [OpBackend.ROCM_AITER, OpBackend.TRITON_GENERIC, OpBackend.PYTORCH_NATIVE],
                 "attn": [OpBackend.TRITON_GENERIC, OpBackend.PYTORCH_NATIVE],
                 "grpo_loss": [OpBackend.TRITON_GRPO_LOSS, OpBackend.PYTORCH_GRPO_LOSS],
+                "ratio_kl": [OpBackend.TRITON_RATIO_KL, OpBackend.PYTORCH_RATIO_KL],
             },
             "cpu": {
                 "logp": [OpBackend.PYTORCH_NATIVE],
                 "attn": [OpBackend.PYTORCH_NATIVE],
                 "grpo_loss": [OpBackend.PYTORCH_GRPO_LOSS],
+                "ratio_kl": [OpBackend.PYTORCH_RATIO_KL],
             },
         }
         logger.info(f"KernelRegistry initialized for {device_ctx.device_type}")
diff --git a/tests/test_grpo_loss.py b/tests/test_grpo_loss.py
index 068632bb..cb7ced30 100644
--- a/tests/test_grpo_loss.py
+++ b/tests/test_grpo_loss.py
@@ -5,8 +5,7 @@
 import torch
 
 from rl_engine.kernels.ops.pytorch.loss.grpo_loss import NativeGRPOLossOp
-from rl_engine.kernels.ops.pytorch.loss.logp import NativeLogpOp
-from rl_engine.kernels.ops.triton.triton_grpo_loss import TritonGRPOLossOp
+from rl_engine.kernels.ops.triton.loss.grpo_loss import TritonGRPOLossOp
 from rl_engine.testing import (
     compute_policy_ratio,
     compute_reference_kl,
@@ -47,15 +46,14 @@ def _batch(seed=0, *, device="cpu", valid_density=0.9):
     )
 
 
-def _logits_like(batch, seed, device="cpu"):
+def _logits(batch, seed, *, device="cpu"):
     gen = torch.Generator(device=device).manual_seed(seed)
     return torch.randn(batch.batch_size, batch.completion_len, _VOCAB, generator=gen, device=device)
 
 
-def _current_logps(batch, seed, *, device="cpu"):
-    """A realistic per-token current-policy logp derived from synthetic logits."""
-    logits = _logits_like(batch, seed, device=device)
-    return selected_logprobs_reference(logits, batch.token_ids)
+def _logit_pair(batch, seed, *, device="cpu"):
+    """(policy_logits, ref_logits) for a batch."""
+    return _logits(batch, seed, device=device), _logits(batch, seed + 1, device=device)
 
 
 def _reference_group_advantages(rewards, samples_per_prompt, eps=1e-6):
@@ -66,19 +64,32 @@ def _reference_group_advantages(rewards, samples_per_prompt, eps=1e-6):
     return ((grouped - group_mean) / group_std).reshape(-1)
 
 
-def _reference_loss(current_logps, old_logps, ref_logps, advantages, mask, clip_eps, beta):
-    """Mirror of examples.grpo_single_gpu.grpo_loss using the testing helpers."""
-    ratio = compute_policy_ratio(current_logps, old_logps, mask)
+def _reference_loss(batch, policy_logits, ref_logits, advantages, clip_eps, beta):
+    """Independent reference: logits -> selected logp -> clipped surrogate + KL."""
+    current = selected_logprobs_reference(policy_logits, batch.token_ids).float()
+    ref = selected_logprobs_reference(ref_logits, batch.token_ids).float()
+    mask = batch.completion_mask
+    ratio = compute_policy_ratio(current, batch.old_logps, mask)
     unclipped = ratio * advantages.float()
     clipped = torch.clamp(ratio, 1.0 - clip_eps, 1.0 + clip_eps) * advantages.float()
     policy_loss_terms = -torch.minimum(unclipped, clipped)
-    kl_terms = compute_reference_kl(current_logps, ref_logps, mask)
+    kl_terms = compute_reference_kl(current, ref, mask)
     policy_loss = masked_mean(policy_loss_terms, mask)
     kl = masked_mean(kl_terms, mask)
     return policy_loss + beta * kl, policy_loss, kl
 
 
-# pure-PyTorch reference op
+def _adv_tokens(batch):
+    sample_adv = _reference_group_advantages(batch.rewards, _SPP)
+    return (
+        sample_adv[:, None]
+        .expand_as(batch.completion_mask)
+        .clone()
+        .masked_fill(~batch.completion_mask, 0.0)
+    )
+
+
+# pure-PyTorch reference op (group advantages)
 def test_group_advantages_matches_reference_uniform():
     op = NativeGRPOLossOp()
     rewards = _batch(seed=1).rewards
@@ -100,24 +111,33 @@ def test_variable_group_boundaries():
     op = NativeGRPOLossOp()
     rewards = torch.tensor([1.0, 3.0, 10.0, 20.0, 30.0])
     got = op.group_advantages(rewards, group_boundaries=[0, 2, 5])
-    # Group 0: [1, 3] -> mean 2, std 1 -> [-1, 1]
     g0 = torch.tensor([-1.0, 1.0])
-    # Group 1: [10, 20, 30] -> mean 20, std sqrt(200/3)
     g1 = (torch.tensor([10.0, 20.0, 30.0]) - 20.0) / (200.0 / 3.0) ** 0.5
     expected = torch.cat([g0, g1])
     assert torch.allclose(got, expected, atol=1e-5)
 
 
+def test_requires_exactly_one_group_spec():
+    op = NativeGRPOLossOp()
+    rewards = _batch(seed=6).rewards
+    with pytest.raises(ValueError):
+        op.group_advantages(rewards)
+    with pytest.raises(ValueError):
+        op.group_advantages(rewards, samples_per_prompt=_SPP, group_boundaries=[0, 4, 8, 12])
+
+
+# pure-PyTorch reference op (loss from logits)
 def test_forward_loss_matches_reference():
     op = NativeGRPOLossOp()
     batch = _batch(seed=0)
-    current = _current_logps(batch, seed=100)
+    policy_logits, ref_logits = _logit_pair(batch, seed=100)
     clip_eps, beta = 0.2, 0.01
 
     loss, policy_loss, kl = op.forward(
-        current,
+        policy_logits,
+        ref_logits,
+        batch.token_ids,
         batch.old_logps,
-        batch.ref_logps,
         batch.rewards,
         batch.completion_mask,
         clip_eps=clip_eps,
@@ -125,31 +145,26 @@ def test_forward_loss_matches_reference():
         samples_per_prompt=_SPP,
     )
 
-    sample_adv = _reference_group_advantages(batch.rewards, samples_per_prompt=_SPP)
-    adv_tokens = (
-        sample_adv[:, None]
-        .expand_as(batch.completion_mask)
-        .clone()
-        .masked_fill(~batch.completion_mask, 0.0)
-    )
     exp_loss, exp_policy, exp_kl = _reference_loss(
-        current, batch.old_logps, batch.ref_logps, adv_tokens, batch.completion_mask, clip_eps, beta
+        batch, policy_logits, ref_logits, _adv_tokens(batch), clip_eps, beta
     )
-
-    assert torch.allclose(loss, exp_loss, atol=1e-6)
-    assert torch.allclose(policy_loss, exp_policy, atol=1e-6)
-    assert torch.allclose(kl, exp_kl, atol=1e-6)
+    assert torch.allclose(loss, exp_loss, atol=1e-5)
+    assert torch.allclose(policy_loss, exp_policy, atol=1e-5)
+    assert torch.allclose(kl, exp_kl, atol=1e-5)
 
 
 def test_gradient_flows_to_policy_logits():
     op = NativeGRPOLossOp()
     batch = _batch(seed=4)
-    current = _current_logps(batch, seed=104).clone().requires_grad_(True)
+    policy_logits, ref_logits = _logit_pair(batch, seed=104)
+    policy_logits = policy_logits.clone().requires_grad_(True)
+    ref_logits = ref_logits.clone().requires_grad_(True)
 
     loss, _, _ = op.forward(
-        current,
+        policy_logits,
+        ref_logits,
+        batch.token_ids,
         batch.old_logps,
-        batch.ref_logps,
         batch.rewards,
         batch.completion_mask,
         clip_eps=0.2,
@@ -158,19 +173,12 @@ def test_gradient_flows_to_policy_logits():
     )
     loss.backward()
 
-    assert current.grad is not None
-    assert torch.isfinite(current.grad).all()
-    # Masked-out tokens must receive zero gradient.
-    assert torch.all(current.grad[~batch.completion_mask] == 0.0)
-
-
-def test_requires_exactly_one_group_spec():
-    op = NativeGRPOLossOp()
-    rewards = _batch(seed=6).rewards
-    with pytest.raises(ValueError):
-        op.group_advantages(rewards)
-    with pytest.raises(ValueError):
-        op.group_advantages(rewards, samples_per_prompt=_SPP, group_boundaries=[0, 4, 8, 12])
+    assert policy_logits.grad is not None
+    assert torch.isfinite(policy_logits.grad).all()
+    # Reference is frozen: no gradient should reach ref_logits.
+    assert ref_logits.grad is None
+    # Masked-out tokens (whole logits rows) must receive zero gradient.
+    assert torch.all(policy_logits.grad[~batch.completion_mask.bool()] == 0.0)
 
 
 # Triton fused op (validated against the native reference)
@@ -179,10 +187,17 @@ def test_triton_forward_matches_native():
     native = NativeGRPOLossOp()
     fused = TritonGRPOLossOp()
     batch = _batch(seed=0, device="cuda")
-    current = _current_logps(batch, seed=100, device="cuda")
+    policy_logits, ref_logits = _logit_pair(batch, seed=100, device="cuda")
+    args = (
+        policy_logits,
+        ref_logits,
+        batch.token_ids,
+        batch.old_logps,
+        batch.rewards,
+        batch.completion_mask,
+    )
     kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
 
-    args = (current, batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask)
     n_loss, n_policy, n_kl = native.forward(*args, **kwargs)
     t_loss, t_policy, t_kl = fused.forward(*args, **kwargs)
 
@@ -196,41 +211,37 @@ def test_triton_backward_matches_native():
     native = NativeGRPOLossOp()
     fused = TritonGRPOLossOp()
     batch = _batch(seed=7, device="cuda")
-    current = _current_logps(batch, seed=107, device="cuda")
-    rest = (batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask)
+    policy_logits, ref_logits = _logit_pair(batch, seed=107, device="cuda")
+    rest = (ref_logits, batch.token_ids, batch.old_logps, batch.rewards, batch.completion_mask)
     kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
 
-    cur_n = current.clone().requires_grad_(True)
-    loss_n, _, _ = native.forward(cur_n, *rest, **kwargs)
-    loss_n.backward()
+    pol_n = policy_logits.clone().requires_grad_(True)
+    native.forward(pol_n, *rest, **kwargs)[0].backward()
 
-    cur_t = current.clone().requires_grad_(True)
-    loss_t, _, _ = fused.forward(cur_t, *rest, **kwargs)
-    loss_t.backward()
+    pol_t = policy_logits.clone().requires_grad_(True)
+    fused.forward(pol_t, *rest, **kwargs)[0].backward()
 
-    assert cur_t.grad is not None
-    assert torch.allclose(cur_t.grad, cur_n.grad, atol=1e-4, rtol=1e-4)
-    assert torch.all(cur_t.grad[~batch.completion_mask] == 0.0)
+    assert pol_t.grad is not None
+    assert torch.allclose(pol_t.grad, pol_n.grad, atol=1e-4, rtol=1e-4)
+    assert torch.all(pol_t.grad[~batch.completion_mask.bool()] == 0.0)
 
 
 @requires_triton_cuda
 def test_triton_backward_with_grad_scaling():
-    """A non-unit upstream gradient must scale the policy-logp gradient linearly."""
+    """A non-unit upstream gradient must scale the policy-logits gradient linearly."""
     fused = TritonGRPOLossOp()
     batch = _batch(seed=3, device="cuda")
-    current = _current_logps(batch, seed=103, device="cuda")
-    rest = (batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask)
+    policy_logits, ref_logits = _logit_pair(batch, seed=103, device="cuda")
+    rest = (ref_logits, batch.token_ids, batch.old_logps, batch.rewards, batch.completion_mask)
     kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
 
-    cur1 = current.clone().requires_grad_(True)
-    loss1, _, _ = fused.forward(cur1, *rest, **kwargs)
-    loss1.backward()
+    pol1 = policy_logits.clone().requires_grad_(True)
+    fused.forward(pol1, *rest, **kwargs)[0].backward()
 
-    cur2 = current.clone().requires_grad_(True)
-    loss2, _, _ = fused.forward(cur2, *rest, **kwargs)
-    (3.0 * loss2).backward()
+    pol2 = policy_logits.clone().requires_grad_(True)
+    (3.0 * fused.forward(pol2, *rest, **kwargs)[0]).backward()
 
-    assert torch.allclose(cur2.grad, 3.0 * cur1.grad, atol=1e-4, rtol=1e-4)
+    assert torch.allclose(pol2.grad, 3.0 * pol1.grad, atol=1e-4, rtol=1e-4)
 
 
 @requires_triton_cuda
@@ -241,7 +252,6 @@ def test_triton_group_advantages_matches_native():
     got = fused.group_advantages(rewards, samples_per_prompt=_SPP)
     expected = native.group_advantages(rewards, samples_per_prompt=_SPP)
     assert torch.allclose(got, expected, atol=1e-5)
-    # Variable group sizes via boundaries.
     got_b = fused.group_advantages(rewards, group_boundaries=[0, 5, 12])
     exp_b = native.group_advantages(rewards, group_boundaries=[0, 5, 12])
     assert torch.allclose(got_b, exp_b, atol=1e-5)
@@ -252,119 +262,40 @@ def test_triton_apply_with_per_sequence_advantages_matches_native():
     native = NativeGRPOLossOp()
     fused = TritonGRPOLossOp()
     batch = _batch(seed=11, device="cuda")
-    current = _current_logps(batch, seed=111, device="cuda")
-
-    sample_adv = native.group_advantages(batch.rewards, samples_per_prompt=_SPP)  # per-sequence
-    adv_tokens = native.expand_advantages(sample_adv, batch.completion_mask)  # per-token for native
-
-    n_loss, _, _ = native.apply(
-        current,
+    policy_logits, ref_logits = _logit_pair(batch, seed=111, device="cuda")
+    sample_adv = native.group_advantages(batch.rewards, samples_per_prompt=_SPP)
+    args = (
+        policy_logits,
+        ref_logits,
+        batch.token_ids,
         batch.old_logps,
-        batch.ref_logps,
-        adv_tokens,
-        batch.completion_mask,
-        clip_eps=0.2,
-        beta=0.05,
-    )
-    t_loss, _, _ = fused.apply(
-        current,
-        batch.old_logps,
-        batch.ref_logps,
         sample_adv,
         batch.completion_mask,
-        clip_eps=0.2,
-        beta=0.05,
-    )
-    assert torch.allclose(t_loss, n_loss, atol=1e-4, rtol=1e-4)
-
-
-# Pipeline composition: logp op -> grpo_loss op
-def test_native_logp_composes_with_native_grpo():
-    logp_op = NativeLogpOp()
-    grpo = NativeGRPOLossOp()
-    batch = _batch(seed=21)
-    logits = _logits_like(batch, seed=121)
-    kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
-    rest = (batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask)
-
-    current = logp_op.apply_fp32(logits, batch.token_ids)
-    loss, _, _ = grpo.forward(current, *rest, **kwargs)
-
-    oracle = selected_logprobs_reference(logits, batch.token_ids)
-    exp, _, _ = grpo.forward(oracle, *rest, **kwargs)
-    assert torch.isfinite(loss)
-    assert torch.allclose(loss, exp, atol=1e-6)
-
-
-def test_native_logp_grpo_pipeline_is_differentiable_to_logits():
-    """NativeLogpOp uses log_softmax/gather, so grads flow logits -> loss."""
-    logp_op = NativeLogpOp()
-    grpo = NativeGRPOLossOp()
-    batch = _batch(seed=22)
-    logits = _logits_like(batch, seed=122).clone().requires_grad_(True)
-
-    current = logp_op.apply_fp32(logits, batch.token_ids)
-    loss, _, _ = grpo.forward(
-        current,
-        batch.old_logps,
-        batch.ref_logps,
-        batch.rewards,
-        batch.completion_mask,
-        clip_eps=0.2,
-        beta=0.05,
-        samples_per_prompt=_SPP,
     )
-    loss.backward()
-
-    assert logits.grad is not None
-    assert torch.isfinite(logits.grad).all()
+    kwargs = dict(clip_eps=0.2, beta=0.05)
 
-
-@requires_triton_cuda
-def test_dispatched_logp_composes_with_triton_grpo():
-    """Real pipeline: dispatched CUDA fused logp -> Triton GRPO loss."""
-    from rl_engine.kernels.registry import kernel_registry
-
-    logp_op = kernel_registry.get_op("logp")
-    grpo = TritonGRPOLossOp()
-    batch = _batch(seed=23, device="cuda")
-    logits = _logits_like(batch, seed=123, device="cuda")
-    kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
-    rest = (batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask)
-
-    current = logp_op.apply_fp32(logits, batch.token_ids)
-    loss, _, _ = grpo.forward(current, *rest, **kwargs)
-
-    oracle = selected_logprobs_reference(logits, batch.token_ids)
-    exp, _, _ = NativeGRPOLossOp().forward(oracle, *rest, **kwargs)
-    assert torch.isfinite(loss)
-    assert torch.allclose(loss, exp, atol=1e-3, rtol=1e-3)
+    n_loss, _, _ = native.apply(*args, **kwargs)
+    t_loss, _, _ = fused.apply(*args, **kwargs)
+    assert torch.allclose(t_loss, n_loss, atol=1e-4, rtol=1e-4)
 
 
 # Loss step: masked-token invariance and a gradient step
-def _perturb_inactive(batch, current):
-    """Set garbage at masked positions; the loss must ignore them."""
-    inactive = ~batch.completion_mask
-    cur = current.clone()
-    old = batch.old_logps.clone()
-    ref = batch.ref_logps.clone()
-    cur[inactive] = 1000.0
-    old[inactive] = -1000.0
-    ref[inactive] = 500.0
-    return cur, old, ref
+def _perturb_inactive_logits(batch, policy_logits):
+    """Set garbage at masked positions' logits; the loss must ignore them."""
+    pol = policy_logits.clone()
+    pol[~batch.completion_mask.bool()] = 1000.0
+    return pol
 
 
 def test_masked_tokens_do_not_affect_native_loss():
     op = NativeGRPOLossOp()
     batch = _batch(seed=8, valid_density=0.75)
-    current = _current_logps(batch, seed=108)
+    policy_logits, ref_logits = _logit_pair(batch, seed=108)
+    args = (ref_logits, batch.token_ids, batch.old_logps, batch.rewards, batch.completion_mask)
     kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
 
-    base, _, _ = op.forward(
-        current, batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask, **kwargs
-    )
-    cur_p, old_p, ref_p = _perturb_inactive(batch, current)
-    pert, _, _ = op.forward(cur_p, old_p, ref_p, batch.rewards, batch.completion_mask, **kwargs)
+    base, _, _ = op.forward(policy_logits, *args, **kwargs)
+    pert, _, _ = op.forward(_perturb_inactive_logits(batch, policy_logits), *args, **kwargs)
     assert torch.allclose(base, pert)
 
 
@@ -372,22 +303,20 @@ def test_masked_tokens_do_not_affect_native_loss():
 def test_masked_tokens_do_not_affect_triton_loss():
     fused = TritonGRPOLossOp()
     batch = _batch(seed=8, device="cuda", valid_density=0.75)
-    current = _current_logps(batch, seed=108, device="cuda")
+    policy_logits, ref_logits = _logit_pair(batch, seed=108, device="cuda")
+    args = (ref_logits, batch.token_ids, batch.old_logps, batch.rewards, batch.completion_mask)
     kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
 
-    base, _, _ = fused.forward(
-        current, batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask, **kwargs
-    )
-    cur_p, old_p, ref_p = _perturb_inactive(batch, current)
-    pert, _, _ = fused.forward(cur_p, old_p, ref_p, batch.rewards, batch.completion_mask, **kwargs)
+    base, _, _ = fused.forward(policy_logits, *args, **kwargs)
+    pert, _, _ = fused.forward(_perturb_inactive_logits(batch, policy_logits), *args, **kwargs)
     assert torch.allclose(base, pert, atol=1e-5)
 
 
-def _descend(op, batch, seed, *, device="cpu", steps=5, lr=0.05):
-    rest = (batch.old_logps, batch.ref_logps, batch.rewards, batch.completion_mask)
+def _descend(op, batch, policy_logits, ref_logits, *, steps=5, lr=0.05):
+    rest = (ref_logits, batch.token_ids, batch.old_logps, batch.rewards, batch.completion_mask)
     kwargs = dict(clip_eps=0.2, beta=0.05, samples_per_prompt=_SPP)
-    initial = op.forward(_current_logps(batch, seed, device=device), *rest, **kwargs)[0]
-    params = _current_logps(batch, seed, device=device).clone().requires_grad_(True)
+    initial = op.forward(policy_logits, *rest, **kwargs)[0]
+    params = policy_logits.clone().requires_grad_(True)
     for _ in range(steps):
         loss, _, _ = op.forward(params, *rest, **kwargs)
         (grad,) = torch.autograd.grad(loss, params)
@@ -397,16 +326,20 @@ def _descend(op, batch, seed, *, device="cpu", steps=5, lr=0.05):
 
 
 def test_grpo_gradient_step_reduces_loss():
-    """Full loss step: forward -> backward -> SGD on the policy logps lowers the loss."""
+    """Full loss step: forward -> backward -> SGD on the policy logits lowers the loss."""
     op = NativeGRPOLossOp()
-    initial, final = _descend(op, _batch(seed=9), seed=109)
+    batch = _batch(seed=9)
+    policy_logits, ref_logits = _logit_pair(batch, seed=109)
+    initial, final = _descend(op, batch, policy_logits, ref_logits)
     assert final.item() < initial.item()
 
 
 @requires_triton_cuda
 def test_triton_grpo_gradient_step_reduces_loss():
     fused = TritonGRPOLossOp()
-    initial, final = _descend(fused, _batch(seed=9, device="cuda"), seed=109, device="cuda")
+    batch = _batch(seed=9, device="cuda")
+    policy_logits, ref_logits = _logit_pair(batch, seed=109, device="cuda")
+    initial, final = _descend(fused, batch, policy_logits, ref_logits)
     assert final.item() < initial.item()
 
 
diff --git a/tests/test_ratio_kl.py b/tests/test_ratio_kl.py
new file mode 100644
index 00000000..1bc9eb0d
--- /dev/null
+++ b/tests/test_ratio_kl.py
@@ -0,0 +1,227 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026 RL-Kernel Contributors
+
+import pytest
+import torch
+
+from rl_engine.kernels.ops.pytorch.loss.ratio_kl import NativeRatioKLOp
+from rl_engine.kernels.ops.triton.loss.ratio_kl import TritonRatioKLOp
+from rl_engine.testing import make_synthetic_rl_kernel_batch, selected_logprobs_reference
+
+try:
+    import triton  # noqa: F401
+
+    _HAS_TRITON = True
+except ImportError:  # pragma: no cover
+    _HAS_TRITON = False
+
+requires_triton_cuda = pytest.mark.skipif(
+    not (_HAS_TRITON and torch.cuda.is_available()),
+    reason="Triton ratio/KL op requires a CUDA device and Triton.",
+)
+
+_NUM_PROMPTS = 3
+_SPP = 4
+_COMP_LEN = 6
+_VOCAB = 64
+
+
+# Shared helpers
+def _batch(seed=0, *, device="cpu", valid_density=0.9):
+    return make_synthetic_rl_kernel_batch(
+        num_prompts=_NUM_PROMPTS,
+        samples_per_prompt=_SPP,
+        prompt_len=0,
+        completion_len=_COMP_LEN,
+        vocab_size=_VOCAB,
+        valid_density=valid_density,
+        device=device,
+        seed=seed,
+    )
+
+
+def _logits(batch, seed, *, vocab=_VOCAB, device="cpu"):
+    gen = torch.Generator(device=device).manual_seed(seed)
+    return torch.randn(batch.batch_size, batch.completion_len, vocab, generator=gen, device=device)
+
+
+def _inputs(seed, *, device="cpu", valid_density=0.9, vocab=_VOCAB):
+    """A full ratio/KL input set: (policy_logits, ref_logits, action_ids, mask, old_logps)."""
+    batch = _batch(seed=seed, device=device, valid_density=valid_density)
+    policy_logits = _logits(batch, seed=seed + 100, vocab=vocab, device=device)
+    ref_logits = _logits(batch, seed=seed + 200, vocab=vocab, device=device)
+    return (
+        policy_logits,
+        ref_logits,
+        batch.token_ids,
+        batch.completion_mask,
+        batch.old_logps,
+    )
+
+
+def _reference_ratio_kl(policy_logits, ref_logits, action_ids, mask, old_logps):
+    """Independent reference using the testing log-prob helper + mask-before-exp."""
+    logp_policy = selected_logprobs_reference(policy_logits, action_ids).float()
+    logp_ref = selected_logprobs_reference(ref_logits, action_ids).float()
+    bool_mask = mask.to(torch.bool)
+    delta = (logp_policy - old_logps.float()).masked_fill(~bool_mask, 0.0)
+    diff = (logp_ref - logp_policy).masked_fill(~bool_mask, 0.0)
+    return torch.exp(delta), torch.exp(diff) - diff - 1.0
+
+
+# pure-PyTorch reference op
+def test_native_matches_reference():
+    op = NativeRatioKLOp()
+    inputs = _inputs(seed=0)
+    ratio, kl = op(*inputs)
+    exp_ratio, exp_kl = _reference_ratio_kl(*inputs)
+    assert torch.allclose(ratio, exp_ratio, atol=1e-6)
+    assert torch.allclose(kl, exp_kl, atol=1e-6)
+
+
+def test_native_masked_tokens_are_neutral():
+    op = NativeRatioKLOp()
+    *_, mask, _ = inputs = _inputs(seed=1, valid_density=0.6)
+    ratio, kl = op(*inputs)
+    inactive = ~mask.to(torch.bool)
+    # mask-before-exp convention: ratio = exp(0) = 1, kl = 0 on inactive tokens.
+    assert torch.allclose(ratio[inactive], torch.ones_like(ratio[inactive]))
+    assert torch.all(kl[inactive] == 0.0)
+
+
+def test_native_ratio_is_one_when_old_equals_policy():
+    op = NativeRatioKLOp()
+    policy_logits, ref_logits, action_ids, mask, _ = _inputs(seed=2, valid_density=1.0)
+    old = selected_logprobs_reference(policy_logits, action_ids).float()
+    ratio, _ = op(policy_logits, ref_logits, action_ids, mask, old)
+    assert torch.allclose(ratio, torch.ones_like(ratio), atol=1e-5)
+
+
+def test_native_gradient_flows_to_policy_logits():
+    op = NativeRatioKLOp()
+    policy_logits, ref_logits, action_ids, mask, old = _inputs(seed=3)
+    policy_logits = policy_logits.clone().requires_grad_(True)
+    ref_logits = ref_logits.clone().requires_grad_(True)
+
+    ratio, kl = op(policy_logits, ref_logits, action_ids, mask, old)
+    (ratio.sum() + kl.sum()).backward()
+
+    assert policy_logits.grad is not None
+    assert torch.isfinite(policy_logits.grad).all()
+    # Reference is frozen: no gradient should reach ref_logits.
+    assert ref_logits.grad is None
+
+
+# Triton fused op (validated against the native reference)
+@requires_triton_cuda
+@pytest.mark.parametrize("vocab", [_VOCAB, 50257])
+def test_triton_forward_matches_native(vocab):
+    native = NativeRatioKLOp()
+    fused = TritonRatioKLOp()
+    inputs = _inputs(seed=4, device="cuda", vocab=vocab)
+    r_t, k_t = fused(*inputs)
+    r_n, k_n = native(*inputs)
+    assert torch.allclose(r_t, r_n, atol=1e-4, rtol=1e-4)
+    assert torch.allclose(k_t, k_n, atol=1e-4, rtol=1e-4)
+
+
+@requires_triton_cuda
+def test_triton_backward_matches_native():
+    native = NativeRatioKLOp()
+    fused = TritonRatioKLOp()
+    policy_logits, ref_logits, action_ids, mask, old = _inputs(seed=5, device="cuda")
+    gr = torch.randn(policy_logits.shape[:-1], device="cuda")
+    gk = torch.randn(policy_logits.shape[:-1], device="cuda")
+
+    pol_t = policy_logits.clone().requires_grad_(True)
+    r_t, k_t = fused(pol_t, ref_logits, action_ids, mask, old)
+    (r_t * gr + k_t * gk).sum().backward()
+
+    pol_n = policy_logits.clone().requires_grad_(True)
+    r_n, k_n = native(pol_n, ref_logits, action_ids, mask, old)
+    (r_n * gr + k_n * gk).sum().backward()
+
+    assert pol_t.grad is not None
+    assert torch.isfinite(pol_t.grad).all()
+    assert torch.allclose(pol_t.grad, pol_n.grad, atol=1e-4, rtol=1e-4)
+
+
+@requires_triton_cuda
+def test_triton_no_grad_to_ref():
+    """The reference is frozen: the fused backward must not reach ref_logits."""
+    fused = TritonRatioKLOp()
+    pol, ref, act, mask, old = _inputs(seed=8, device="cuda")
+    pol = pol.clone().requires_grad_(True)
+    ref = ref.clone().requires_grad_(True)
+    r, k = fused(pol, ref, act, mask, old)
+    (r.sum() + k.sum()).backward()
+    assert pol.grad is not None
+    assert ref.grad is None
+
+
+@requires_triton_cuda
+def test_triton_backward_with_grad_scaling():
+    """A non-unit upstream gradient must scale the policy-logits gradient linearly."""
+    fused = TritonRatioKLOp()
+    policy_logits, ref_logits, action_ids, mask, old = _inputs(seed=6, device="cuda")
+
+    pol1 = policy_logits.clone().requires_grad_(True)
+    r1, k1 = fused(pol1, ref_logits, action_ids, mask, old)
+    (r1.sum() + k1.sum()).backward()
+
+    pol2 = policy_logits.clone().requires_grad_(True)
+    r2, k2 = fused(pol2, ref_logits, action_ids, mask, old)
+    (3.0 * (r2.sum() + k2.sum())).backward()
+
+    assert torch.allclose(pol2.grad, 3.0 * pol1.grad, atol=1e-4, rtol=1e-4)
+
+
+@requires_triton_cuda
+def test_triton_masked_tokens_do_not_affect_active():
+    """Garbage logits at masked positions must not change active outputs."""
+    fused = TritonRatioKLOp()
+    policy_logits, ref_logits, action_ids, mask, old = _inputs(
+        seed=7, device="cuda", valid_density=0.7
+    )
+    base_r, base_k = fused(policy_logits, ref_logits, action_ids, mask, old)
+
+    inactive = ~mask.to(torch.bool)
+    pert = policy_logits.clone()
+    pert[inactive] = 1000.0
+    pert_r, pert_k = fused(pert, ref_logits, action_ids, mask, old)
+
+    active = mask.to(torch.bool)
+    assert torch.allclose(base_r[active], pert_r[active], atol=1e-5)
+    assert torch.allclose(base_k[active], pert_k[active], atol=1e-5)
+    assert torch.allclose(pert_r[inactive], torch.ones_like(pert_r[inactive]))
+
+
+@requires_triton_cuda
+def test_triton_handles_oob_action_ids():
+    """Out-of-range ids at masked positions must not fault; the fused op clamps
+    them (like the native op) and stays finite, agreeing on active outputs."""
+    native = NativeRatioKLOp()
+    fused = TritonRatioKLOp()
+    pol, ref, act, mask, old = _inputs(seed=9, device="cuda", valid_density=0.7)
+    act = act.clone()
+    inactive = ~mask.to(torch.bool)
+    act[inactive] = _VOCAB + 999  # garbage id at masked positions
+
+    r_t, k_t = fused(pol, ref, act, mask, old)
+    r_n, k_n = native(pol, ref, act, mask, old)
+
+    active = mask.to(torch.bool)
+    assert torch.isfinite(r_t).all() and torch.isfinite(k_t).all()
+    assert torch.allclose(r_t[active], r_n[active], atol=1e-4, rtol=1e-4)
+    assert torch.allclose(k_t[active], k_n[active], atol=1e-4, rtol=1e-4)
+
+
+# Registry dispatch (device-dependent backend selection)
+def test_registry_dispatches_ratio_kl():
+    from rl_engine.kernels.registry import kernel_registry
+
+    op = kernel_registry.get_op("ratio_kl")
+    if _HAS_TRITON and torch.cuda.is_available():
+        assert isinstance(op, TritonRatioKLOp)
+    else:
+        assert isinstance(op, NativeRatioKLOp)