diff --git a/docs/source/features/sampling.md b/docs/source/features/sampling.md
index 277aa64eeddc..053d2720efd8 100644
--- a/docs/source/features/sampling.md
+++ b/docs/source/features/sampling.md
@@ -196,8 +196,10 @@ modes.
     `repetition_penalty`. Values `<= 0` have no effect, and values larger than the prompt
     are clamped to the prompt length.
 
-  * Occurrence penalties are not supported in combination with beam search; such requests
-    are rejected.
+  * With beam search the occurrence history is kept per beam rather than per request:
+    each beam is penalized against the tokens on its own path, and whenever a beam
+    continues another one it inherits that beam's history. The prompt seeds every beam
+    alike, so `prompt_ignore_length` applies to all of them equally.
 
 * If `no_repeat_ngram_size = n` is specified, any token that would recreate an `n`-gram already
   present in the sequence (prompt included) is excluded from sampling. `None` or `0` disables
diff --git a/tensorrt_llm/_torch/pyexecutor/sampler/ops/vanilla.py b/tensorrt_llm/_torch/pyexecutor/sampler/ops/vanilla.py
index 88b682813c8d..ee093a714980 100644
--- a/tensorrt_llm/_torch/pyexecutor/sampler/ops/vanilla.py
+++ b/tensorrt_llm/_torch/pyexecutor/sampler/ops/vanilla.py
@@ -379,8 +379,148 @@ def update_occurrence_workspace(
             # Marking a dense bool mask is idempotent, so duplicate tokens are safe.
             presence_prefix_cuda[prefix_slots, prefix_tokens] = True
 
-    # fullgraph=True is safe here: served model has fixed shapes and compiles ~2 graphs,
-    # well under the default limit (8)
+    # --- Beam-search occurrence counts --------------------------------------
+    # Counterpart of the per-beam workspace handling in the C++ ``batchApplyPenalty``
+    # kernel (``penaltyKernels.cu:151-171``): a beam does not re-walk its history,
+    # it inherits its parent beam's counts and appends the single token it just
+    # emitted. ``counts_cuda`` is flat ``[num_slots * max_beam_width, vocab_size]``;
+    # beam ``b`` of slot ``s`` owns row ``s * max_beam_width + b``, which collapses to
+    # plain slot indexing at ``max_beam_width == 1``. A single-beam engine never calls
+    # this op -- it folds inside the penalty graph instead.
+
+    @staticmethod
+    @torch.compile(fullgraph=True, mode="max-autotune-no-cudagraphs")
+    def _update_beam_occurrence_counts_impl(
+        counts_cuda: torch.Tensor,
+        active_cuda: torch.Tensor,
+        has_previous_token_cuda: torch.Tensor,
+        beam_slot_cuda: torch.Tensor,
+        new_tokens: torch.Tensor,
+        predecessor_beams: torch.Tensor,
+        seq_slots: torch.Tensor,
+        request_num_beams: torch.Tensor,
+        max_beam_width: int,
+    ) -> None:
+        """Re-parent every beam onto the beam it continues, then fold in its token.
+
+        Beam ``b`` of slot ``s`` takes over the counts of ``predecessor_beams[s, b]``
+        before this step's token is folded in. Slots that did not sample last step read
+        the identity permutation instead, gated by ``has_previous_token``; single-beam
+        slots on a beam engine are gated out by ``beam_slot_cuda``, since their
+        ``predecessor_beams`` row is never written.
+
+        ``armed`` gates per *slot* (``.unsqueeze(1)``), so beams past the current width are
+        re-parented too, from a clamped and possibly stale parent -- safe because nothing
+        reads such a row until the width grows to cover it, and growth re-parents it from a
+        beam that was valid last step. The gate is device-side, so this runs
+        unconditionally rather than pay a D2H sync.
+
+        NB: ``fullgraph=True``, mutates ``counts_cuda`` in place, and every batch-varying
+        dim-0 argument must be marked dynamic by the caller -- an unmarked peer forces the
+        marked dims to specialize. Add such an argument only together with its
+        ``mark_dynamic`` in ``update_beam_occurrence_counts``.
+        """
+        vocab = counts_cuda.size(-1)
+        beam_ids = torch.arange(max_beam_width, device=counts_cuda.device)
+
+        armed = (
+            active_cuda[seq_slots] & has_previous_token_cuda[seq_slots] & beam_slot_cuda[seq_slots]
+        ).unsqueeze(1)
+        # Beams past the current width hold stale parents; clamping keeps the
+        # gather in range. Their rows are never read while masked out, and a later
+        # beam-width growth only ever re-gathers from a valid parent row.
+        parent = predecessor_beams[seq_slots].to(torch.int64).clamp(0, max_beam_width - 1)
+        src_beam = torch.where(armed, parent, beam_ids.expand_as(parent))
+        base = seq_slots.unsqueeze(1) * max_beam_width
+        counts_cuda.index_copy_(
+            0,
+            (base + beam_ids).reshape(-1),
+            counts_cuda.index_select(0, (base + src_beam).reshape(-1)),
+        )
+
+        # Same masked flat scatter as the single-beam fold, fanned out over the
+        # beam axis: masked entries add 0 at counts[row, 0], so inactive, unarmed,
+        # out-of-layout and padded-token beams are no-ops. A beam the previous step
+        # did not produce carries BEAM_SEARCH_PAD_TOKEN (-1), which the range check
+        # rejects -- that is what confines the fold to the beams that actually
+        # sampled, since ``request_num_beams`` is the (wider) row-layout width.
+        previous_token = new_tokens[0, seq_slots, :].to(torch.int64)  # [R, max_beam_width]
+        fold_ok = (
+            (active_cuda[seq_slots] & has_previous_token_cuda[seq_slots]).unsqueeze(1)
+            & (beam_ids.unsqueeze(0) < request_num_beams.unsqueeze(1))
+            & (previous_token >= 0)
+            & (previous_token < vocab)
+        )
+        rows = base + beam_ids  # [R, max_beam_width]
+        flat_index = rows * vocab + torch.where(
+            fold_ok, previous_token, previous_token.new_zeros(())
+        )
+        counts_cuda.view(-1).scatter_add_(
+            0, flat_index.reshape(-1), fold_ok.reshape(-1).to(counts_cuda.dtype)
+        )
+
+    @staticmethod
+    def update_beam_occurrence_counts(
+        counts_cuda: torch.Tensor,
+        active_cuda: torch.Tensor,
+        has_previous_token_cuda: torch.Tensor,
+        beam_slot_cuda: torch.Tensor,
+        new_tokens: torch.Tensor,
+        predecessor_beams: torch.Tensor,
+        seq_slots: torch.Tensor,
+        request_num_beams: torch.Tensor,
+        max_beam_width: int,
+    ) -> None:
+        """Advance the per-beam occurrence counts by one step, in place.
+
+        Re-parents every beam onto the beam it continues, then folds in the token
+        each beam sampled last step. Must run before anything reads the counts for
+        this step, and before this step's sampling overwrites ``predecessor_beams``.
+        This wrapper only marks the batch-varying dims dynamic; the work is in
+        ``_update_beam_occurrence_counts_impl``.
+
+        Also covers single-beam requests sharing a beam engine: ``beam_slot_cuda``
+        turns their re-parent into the identity, and ``request_num_beams == 1``
+        confines their fold to beam 0. Context requests are likewise laid out at
+        one beam and read as such.
+
+        Args:
+            counts_cuda: ``int32[num_slots * max_beam_width, vocab_size]`` workspace.
+            active_cuda / has_previous_token_cuda: per-slot gates, length ``num_slots``.
+            new_tokens: ``[max_tokens, num_slots, max_beam_width]`` device buffer
+                holding the previous step's sampled token per beam.
+            predecessor_beams: ``int32[num_slots, max_beam_width]``, the parent beam
+                of each beam as written by the previous step's beam search.
+            seq_slots: ``int64[R]`` slot per request.
+            request_num_beams: ``[R]`` row-layout beam width per request, i.e. the
+                static admission width the logits rows are laid out at (1 for a
+                context request), not the per-iteration width. Beams at or past it
+                are skipped; under a growing ``beam_width_array`` the beams between
+                the per-iteration and the layout width are skipped instead by their
+                BEAM_SEARCH_PAD_TOKEN.
+        """
+        if seq_slots.numel() == 0:
+            return
+        # Batch-varying dim-0 tensors; mark every one, or an unmarked peer forces the
+        # marked dims to specialize (cf. apply_batched_occurrence_penalties). The other
+        # arguments keep dim 0 == num_slots (fixed), so they must NOT be marked.
+        torch._dynamo.mark_dynamic(seq_slots, 0)
+        torch._dynamo.mark_dynamic(request_num_beams, 0)
+        Fusions._update_beam_occurrence_counts_impl(
+            counts_cuda,
+            active_cuda,
+            has_previous_token_cuda,
+            beam_slot_cuda,
+            new_tokens,
+            predecessor_beams,
+            seq_slots,
+            request_num_beams,
+            max_beam_width,
+        )
+
+    # fullgraph=True is safe here: 4 cache entries per process against a default limit of 8
+    # -- the size-1 specialization of the dynamic dim, times the `prefix_seen_cuda` Optional
+    # flipping on first use. The beam parameters are per-engine constants and cost nothing.
     @staticmethod
     @torch.compile(fullgraph=True, mode="max-autotune-no-cudagraphs")
     def _apply_occurrence_penalties_impl(
@@ -396,24 +536,32 @@ def _apply_occurrence_penalties_impl(
         repetition_cuda: torch.Tensor,
         presence_cuda: torch.Tensor,
         frequency_cuda: torch.Tensor,
+        request_num_beams: Optional[torch.Tensor],
+        max_beam_width: int,
+        fold_pending: bool,
     ) -> None:
         vocab = logits.size(-1)
 
         # Fold the device-pending sampled token into the persistent counts, once per armed
         # active slot, before the gather reads them, via one flat scatter_add. Masked entries
         # add 0 at counts[slot, 0], so inactive/unarmed/out-of-range slots are no-ops.
-        previous_token = new_tokens[0, seq_slots, 0].to(torch.int64)
-        fold_ok = (
-            active_cuda[seq_slots]
-            & has_previous_token_cuda[seq_slots]
-            & (request_num_steps > 0)
-            & (previous_token >= 0)
-            & (previous_token < vocab)
-        )
-        flat_index = seq_slots * vocab + torch.where(
-            fold_ok, previous_token, previous_token.new_zeros(())
-        )
-        counts_cuda.view(-1).scatter_add_(0, flat_index, fold_ok.to(counts_cuda.dtype))
+        # Only reached on a single-beam engine, so ``max_beam_width`` is 1 and the row
+        # scaling folds away. A beam engine has to re-parent before folding, which it
+        # cannot do here, so it folds in ``update_beam_occurrence_counts`` instead.
+        if fold_pending:
+            slot_rows = seq_slots * max_beam_width
+            previous_token = new_tokens[0, seq_slots, 0].to(torch.int64)
+            fold_ok = (
+                active_cuda[seq_slots]
+                & has_previous_token_cuda[seq_slots]
+                & (request_num_steps > 0)
+                & (previous_token >= 0)
+                & (previous_token < vocab)
+            )
+            flat_index = slot_rows * vocab + torch.where(
+                fold_ok, previous_token, previous_token.new_zeros(())
+            )
+            counts_cuda.view(-1).scatter_add_(0, flat_index, fold_ok.to(counts_cuda.dtype))
 
         # Map each logits row to its owning request with a broadcasted range comparison.
         # This is O(T * R), but T and R are both small (rows per step x requests) and the
@@ -424,12 +572,29 @@ def _apply_occurrence_penalties_impl(
         # reads back from the device, and that per-step D2H sync destroys the overlap
         # between the sampler's host work and the model forward (measured ~20x slower).
         rows = torch.arange(logits.size(0), device=logits.device).unsqueeze(1)  # [T, 1]
-        owned = (rows >= request_offsets) & (rows < request_offsets + request_num_steps)  # [T, R]
+        # A request owns num_steps * num_beams rows, laid out beam-major / step-minor.
+        # request_num_beams is None on a single-beam engine, collapsing the span to
+        # num_steps and the counts row to plain slot indexing.
+        span = (
+            request_num_steps
+            if request_num_beams is None
+            else request_num_steps * request_num_beams
+        )
+        owned = (rows >= request_offsets) & (rows < request_offsets + span)  # [T, R]
         row_owned = owned.any(dim=1)  # [T]
         row_slot = (owned * seq_slots).sum(dim=1)  # [T]; slot per row, 0 for unowned
         row_active = row_owned & active_cuda[row_slot]
 
-        count = counts_cuda[row_slot]
+        if request_num_beams is None:
+            row_counts = row_slot
+        else:
+            # Recover the beam from the row's offset within its request: beam-major means
+            # beam = local_index // num_steps.
+            local = (owned * (rows - request_offsets)).sum(dim=1)  # [T]
+            steps = (owned * request_num_steps).sum(dim=1).clamp(min=1)  # [T]; avoid //0
+            row_counts = row_slot * max_beam_width + local // steps
+
+        count = counts_cuda[row_counts]
         rep = repetition_cuda[row_slot].unsqueeze(1)
         pre = presence_cuda[row_slot].unsqueeze(1)
         freq = frequency_cuda[row_slot].unsqueeze(1)
@@ -466,14 +631,17 @@ def apply_batched_occurrence_penalties(
         repetition_cuda: torch.Tensor,
         presence_cuda: torch.Tensor,
         frequency_cuda: torch.Tensor,
+        request_num_beams: Optional[torch.Tensor] = None,
+        max_beam_width: int = 1,
+        fold_pending: bool = True,
     ) -> None:
         """Apply occurrence penalties to ``logits`` in place, before temperature handling.
 
         Args:
             logits: ``[T, vocab_size]`` packed generated-token logits, where
                 ``T == sum(num_steps * num_beams)``. Request ``r`` owns the rows
-                ``request_offsets[r] + step`` for ``step in [0, request_num_steps[r])``;
-                rows no request owns are left bit-identical. Modified in place.
+                ``request_offsets[r] + beam * num_steps[r] + step``, i.e. beam-major /
+                step-minor; rows no request owns are left bit-identical. Modified in place.
             counts_cuda / presence_prefix_cuda: the occurrence workspace; see
                 ``PenaltyHandler.PenaltyStore`` for their semantics.
             active_cuda / has_previous_token_cuda / repetition_cuda / presence_cuda /
@@ -484,6 +652,18 @@ def apply_batched_occurrence_penalties(
             request_offsets / request_num_steps: ``[R]`` device tensors, already
                 staged by the caller. The owned spans must not overlap, but they need
                 not be ordered, and rows they skip are left bit-identical.
+            request_num_beams: ``[R]`` row-layout beam width per request -- the static
+                admission width ModelEngine lays the rows out at, which under a growing
+                ``beam_width_array`` exceeds the per-iteration width -- so a row can be
+                mapped back to the beam that owns it. None on a single-beam engine.
+            max_beam_width: the engine's beam width, the stride of the counts rows.
+            fold_pending: whether to fold the device-pending token here. False on a beam
+                engine, where ``update_beam_occurrence_counts`` has already re-parented
+                and folded every slot.
+
+        The last three are compile-time constants to Dynamo (an Optional and two Python
+        scalars), so each engine specializes to its own graph and neither carries the
+        other's branches.
 
         All heavy lifting is fused into the single compiled ``_apply_occurrence_penalties_impl``
         graph; this wrapper only marks the batch-varying dims dynamic.
@@ -498,6 +678,8 @@ def apply_batched_occurrence_penalties(
         torch._dynamo.mark_dynamic(seq_slots, 0)
         torch._dynamo.mark_dynamic(request_offsets, 0)
         torch._dynamo.mark_dynamic(request_num_steps, 0)
+        if request_num_beams is not None:
+            torch._dynamo.mark_dynamic(request_num_beams, 0)
         Fusions._apply_occurrence_penalties_impl(
             logits,
             counts_cuda,
@@ -511,4 +693,7 @@ def apply_batched_occurrence_penalties(
             repetition_cuda,
             presence_cuda,
             frequency_cuda,
+            request_num_beams,
+            max_beam_width,
+            fold_pending,
         )
diff --git a/tensorrt_llm/_torch/pyexecutor/sampler/penalties.py b/tensorrt_llm/_torch/pyexecutor/sampler/penalties.py
index b688a5c696a1..5057376ed70c 100644
--- a/tensorrt_llm/_torch/pyexecutor/sampler/penalties.py
+++ b/tensorrt_llm/_torch/pyexecutor/sampler/penalties.py
@@ -65,9 +65,16 @@ class PenaltyStore:
       other token (the rest of the prompt plus each generated token) increments
       ``counts_cuda``, which drives presence/frequency and -- via ``counts > 0`` --
       repetition as well.
+
+    ``counts_cuda`` carries one row per *beam*: beam ``b`` of slot ``s`` owns row
+    ``s * max_beam_width + b`` (see ``counts_rows``), because beams diverge and each
+    needs its own history. ``presence_prefix_cuda`` stays one row per slot -- the
+    ignored prompt prefix is shared by every beam of a request. With
+    ``max_beam_width == 1`` the beam axis vanishes and both are plain slot-indexed.
     """
 
     max_num_sequences: int
+    max_beam_width: int
     device: torch.device
 
     # --- Penalty parameters (allocateBuffer counterpart), shape [max_num_sequences] ---
@@ -81,10 +88,18 @@ class PenaltyStore:
     """bool[slots]; whether a slot has an active occurrence penalty."""
     has_previous_token_cuda: torch.Tensor
     """bool[slots]; whether ``new_tokens`` contains a token to accumulate."""
+    beam_slot_cuda: torch.Tensor
+    """bool[slots]; whether the slot's counts must be re-parented each step.
+
+    Beam width is a per-request property, so a beam engine may host single-beam requests
+    too (see ``py_executor._validate_request`` / TRTLLM-14792). Only true beam slots have
+    a meaningful ``predecessor_beams`` row; a single-beam slot's is never written and must
+    not be believed. Stays all-False on a single-beam engine."""
 
     # --- Occurrence workspace (allocateWorkspace counterpart), allocated lazily ---
     counts_cuda: torch.Tensor | None = None
-    """int32[slots, vocab_size] or None; occurrence counts (see class docstring)."""
+    """int32[slots * max_beam_width, vocab_size] or None; occurrence counts
+    (see class docstring)."""
     presence_prefix_cuda: torch.Tensor | None = None
     """bool[slots, vocab_size] or None; ignored-prompt-prefix presence mask."""
 
@@ -92,9 +107,13 @@ class PenaltyStore:
     # ``stage_request_metadata`` so the hot path does not allocate per step.
     request_offsets_cuda: torch.Tensor | None = None
     request_num_steps_cuda: torch.Tensor | None = None
+    request_num_beams_cuda: torch.Tensor | None = None
+    """Stays None on a single-beam engine, which never stages a beam width."""
 
     @classmethod
-    def create(cls, *, max_num_sequences: int, device: torch.device) -> "PenaltyStore":
+    def create(
+        cls, *, max_num_sequences: int, max_beam_width: int, device: torch.device
+    ) -> "PenaltyStore":
         """Allocate the vocab-independent buffers with their no-op defaults.
 
         ``inference_mode(False)`` guards every allocation in this class: the
@@ -104,6 +123,7 @@ def create(cls, *, max_num_sequences: int, device: torch.device) -> "PenaltyStor
         with torch.inference_mode(False):
             return cls(
                 max_num_sequences=max_num_sequences,
+                max_beam_width=max_beam_width,
                 device=device,
                 repetition_cuda=torch.ones(max_num_sequences, dtype=torch.float32, device=device),
                 presence_cuda=torch.zeros(max_num_sequences, dtype=torch.float32, device=device),
@@ -112,8 +132,20 @@ def create(cls, *, max_num_sequences: int, device: torch.device) -> "PenaltyStor
                 has_previous_token_cuda=torch.zeros(
                     max_num_sequences, dtype=torch.bool, device=device
                 ),
+                beam_slot_cuda=torch.zeros(max_num_sequences, dtype=torch.bool, device=device),
             )
 
+    def counts_rows(self, slots_cuda: torch.Tensor) -> torch.Tensor:
+        """Map slot indices to the ``counts_cuda`` rows they own.
+
+        One row per beam, so slot ``s`` owns ``s * max_beam_width + b``. Returns
+        ``slots_cuda`` unchanged in the single-beam case, where the two coincide.
+        """
+        if self.max_beam_width == 1:
+            return slots_cuda
+        beams = torch.arange(self.max_beam_width, device=slots_cuda.device)
+        return (slots_cuda.unsqueeze(1) * self.max_beam_width + beams).reshape(-1)
+
     def ensure_workspace(self, *, vocab_size: int, needs_prefix: bool) -> None:
         """Allocate the vocab-sized workspace on first use.
 
@@ -124,7 +156,7 @@ def ensure_workspace(self, *, vocab_size: int, needs_prefix: bool) -> None:
         with torch.inference_mode(False):
             if self.counts_cuda is None:
                 self.counts_cuda = torch.zeros(
-                    (self.max_num_sequences, vocab_size),
+                    (self.max_num_sequences * self.max_beam_width, vocab_size),
                     dtype=torch.int32,
                     device=self.device,
                 )
@@ -136,13 +168,21 @@ def ensure_workspace(self, *, vocab_size: int, needs_prefix: bool) -> None:
                 )
 
     def stage_request_metadata(
-        self, request_offsets_host: torch.Tensor, request_num_steps_host: torch.Tensor
-    ) -> tuple[torch.Tensor, torch.Tensor]:
+        self,
+        request_offsets_host: torch.Tensor,
+        request_num_steps_host: torch.Tensor,
+        request_num_beams_host: torch.Tensor | None = None,
+    ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor | None]:
         """Copy this step's ``[R]`` request metadata into persistent device buffers.
 
-        The host tensors are already pinned by the caller, so each step costs two
-        small async H2D copies into a reused allocation rather than two fresh
-        device tensors. Returned views are only valid until the next call.
+        The host tensors are already pinned by the caller, so each step costs a couple
+        of small async H2D copies into a reused allocation rather than fresh device
+        tensors. Returned views are only valid until the next call.
+
+        All three buffers are ``[R]`` and grow together under one capacity check.
+        ``request_num_beams_host`` is omitted on a single-beam engine, which then gets
+        ``None`` back and allocates no third buffer; a beam engine must pass it on every
+        call, including the first, or the buffer is never allocated.
         """
         num_requests = request_offsets_host.numel()
         with torch.inference_mode(False):
@@ -157,12 +197,24 @@ def stage_request_metadata(
                 self.request_num_steps_cuda = torch.empty(
                     capacity, dtype=request_num_steps_host.dtype, device=self.device
                 )
+                self.request_num_beams_cuda = (
+                    torch.empty(capacity, dtype=request_num_beams_host.dtype, device=self.device)
+                    if request_num_beams_host is not None
+                    else None
+                )
         assert self.request_num_steps_cuda is not None
         offsets = self.request_offsets_cuda[:num_requests]
         num_steps = self.request_num_steps_cuda[:num_requests]
         offsets.copy_(request_offsets_host, non_blocking=True)
         num_steps.copy_(request_num_steps_host, non_blocking=True)
-        return offsets, num_steps
+        if request_num_beams_host is None:
+            return offsets, num_steps, None
+        assert self.request_num_beams_cuda is not None, (
+            "a beam engine must stage request_num_beams from its first call onwards"
+        )
+        num_beams = self.request_num_beams_cuda[:num_requests]
+        num_beams.copy_(request_num_beams_host, non_blocking=True)
+        return offsets, num_steps, num_beams
 
 
 class PenaltyHandler:
@@ -188,15 +240,18 @@ class _SlotState:
         """Per-slot host-only bookkeeping (never read by the ops)."""
 
         prompt_ignore_length: int
+        uses_beam_search: bool = False
         initialized: bool = False
 
     def __init__(
         self,
         *,
         max_num_sequences: int,
+        max_beam_width: int,
         device: torch.device | str,
     ):
         self._max_num_sequences = max_num_sequences
+        self._max_beam_width = max_beam_width
         self._device = torch.device(device)
         # Whether any (past or current) active request uses prompt_ignore_length > 0,
         # which requires allocating the presence-prefix mask.
@@ -210,20 +265,12 @@ def __init__(
         self._new_repetition: list[float] = []
         self._new_presence: list[float] = []
         self._new_frequency: list[float] = []
-        self.store = PenaltyStore.create(max_num_sequences=max_num_sequences, device=self._device)
-
-    @staticmethod
-    def validate_request(request: LlmRequest) -> None:
-        """Reject unsupported combinations for a penalized request.
-
-        Called from ``TorchSampler.validate_request`` (request admission), so a
-        violating request is failed individually instead of aborting the whole batch.
-        """
-        if _get_max_beam_width(request) > 1 and has_occurrence_penalty(request):
-            raise ValueError(
-                "TorchSampler does not support repetition, presence, or frequency "
-                "penalties with beam search."
-            )
+        self._new_beam_slot: list[bool] = []
+        self.store = PenaltyStore.create(
+            max_num_sequences=max_num_sequences,
+            max_beam_width=max_beam_width,
+            device=self._device,
+        )
 
     def _to_device(self, values: list[int], dtype: torch.dtype) -> torch.Tensor:
         return torch.tensor(values, dtype=dtype, pin_memory=prefer_pinned()).to(
@@ -240,7 +287,7 @@ def prepare_for_new_request(self, request: LlmRequest, slot: int) -> None:
         gathered, so their stale parameters/counts are left untouched.
         """
         was_active = self._slots[slot] is not None
-        if not (_get_max_beam_width(request) == 1 and has_occurrence_penalty(request)):
+        if not has_occurrence_penalty(request):
             self._slots[slot] = None
             if was_active:
                 self._num_active_slots -= 1
@@ -259,7 +306,11 @@ def prepare_for_new_request(self, request: LlmRequest, slot: int) -> None:
         if prompt_ignore_length > 0:
             self._needs_prefix = True
 
-        self._slots[slot] = self._SlotState(prompt_ignore_length=prompt_ignore_length)
+        uses_beam_search = self._max_beam_width > 1 and _get_max_beam_width(request) > 1
+        self._slots[slot] = self._SlotState(
+            prompt_ignore_length=prompt_ignore_length,
+            uses_beam_search=uses_beam_search,
+        )
         if not was_active:
             self._num_active_slots += 1
 
@@ -267,6 +318,7 @@ def prepare_for_new_request(self, request: LlmRequest, slot: int) -> None:
         self._new_repetition.append(repetition if repetition is not None else 1.0)
         self._new_presence.append(presence if presence is not None else 0.0)
         self._new_frequency.append(frequency if frequency is not None else 0.0)
+        self._new_beam_slot.append(uses_beam_search)
 
     def update_for_new_requests(self, *, new_seq_slots_cuda_long: torch.Tensor) -> None:
         """Flush this step's admissions to the device in a handful of batched updates.
@@ -279,6 +331,10 @@ def update_for_new_requests(self, *, new_seq_slots_cuda_long: torch.Tensor) -> N
         store = self.store
         store.active_cuda.index_fill_(0, new_seq_slots_cuda_long, False)
         store.has_previous_token_cuda.index_fill_(0, new_seq_slots_cuda_long, False)
+        if self._max_beam_width > 1:
+            # Stays all-False on a single-beam engine, so clearing it there would be a
+            # kernel launch per step for nothing.
+            store.beam_slot_cuda.index_fill_(0, new_seq_slots_cuda_long, False)
 
         if not self._new_slots:
             return
@@ -294,10 +350,19 @@ def update_for_new_requests(self, *, new_seq_slots_cuda_long: torch.Tensor) -> N
         store.presence_cuda.index_copy_(0, slots_cuda, params_cuda[1])
         store.frequency_cuda.index_copy_(0, slots_cuda, params_cuda[2])
         store.active_cuda.index_fill_(0, slots_cuda, True)
+        if self._max_beam_width > 1:
+            store.beam_slot_cuda.index_copy_(
+                0,
+                slots_cuda,
+                torch.tensor(self._new_beam_slot, dtype=torch.bool, pin_memory=prefer_pinned()).to(
+                    self._device, non_blocking=True
+                ),
+            )
 
         # Re-zero the workspace rows so a prior occupant's counts do not leak in.
+        # counts_cuda holds one row per beam, so every beam of the slot must be cleared.
         if store.counts_cuda is not None:
-            store.counts_cuda.index_fill_(0, slots_cuda, 0)
+            store.counts_cuda.index_fill_(0, store.counts_rows(slots_cuda), 0)
         if store.presence_prefix_cuda is not None:
             store.presence_prefix_cuda.index_fill_(0, slots_cuda, False)
 
@@ -305,6 +370,7 @@ def update_for_new_requests(self, *, new_seq_slots_cuda_long: torch.Tensor) -> N
         self._new_repetition.clear()
         self._new_presence.clear()
         self._new_frequency.clear()
+        self._new_beam_slot.clear()
 
     def _initialize_workspace(
         self,
@@ -328,6 +394,7 @@ def _initialize_workspace(
 
         # One conversion for the whole prompt; the split point is just
         # prompt_ignore_length, so the two groups are plain slices.
+        base_row = slot * self._max_beam_width
         tokens = self._to_device(prompt, torch.int64)
         prefix_tokens = tokens[: state.prompt_ignore_length]
         counted_tokens = tokens[state.prompt_ignore_length :]
@@ -341,11 +408,18 @@ def _initialize_workspace(
         Fusions.update_occurrence_workspace(
             counts_cuda,
             self.store.presence_prefix_cuda,
-            torch.full_like(counted_tokens, slot),
+            torch.full_like(counted_tokens, base_row),
             counted_tokens,
+            # The prefix mask is per slot: every beam shares the prompt.
             torch.full_like(prefix_tokens, slot),
             prefix_tokens,
         )
+        if state.uses_beam_search:
+            # Every beam starts from the same prompt. Seeding all of them, rather than
+            # only beam 0, also covers a generation-only (disaggregated decode) request
+            # whose first penalized step already has several beams and hence no
+            # re-parenting to broadcast beam 0's counts for it.
+            counts_cuda[base_row + 1 : base_row + self._max_beam_width].copy_(counts_cuda[base_row])
 
     def update_token_counts(
         self,
@@ -364,6 +438,9 @@ def update_token_counts(
 
         counts_cuda = self.store.counts_cuda
         assert counts_cuda is not None
+        # Only speculative decoding reaches here, and it is rejected together with
+        # beam search (TorchSampler.__init__), so beam 0 is the only live beam.
+        assert self._max_beam_width == 1, "speculative token commit is single-beam only"
         vocab_size = counts_cuda.size(-1)
         consumed_slots: list[int] = []
         counted_slots: list[int] = []
@@ -404,17 +481,35 @@ def apply(
         seq_slots: torch.Tensor,
         request_offsets: torch.Tensor,
         request_num_steps: torch.Tensor,
+        request_num_beams: torch.Tensor | None = None,
+        predecessor_beams: torch.Tensor | None = None,
         is_draft_batch: bool = False,
     ) -> None:
-        """Apply the occurrence penalties to ``logits`` in place.
+        """Advance the occurrence state for this step and apply the penalties to ``logits``.
 
         ``logits`` is the packed generated-token logits ``[sum(num_steps * num_beams),
-        vocab_size]``; request ``r`` owns ``request_num_steps[r]`` consecutive rows
-        starting at ``request_offsets[r]``, in beam-major / step-minor order.
-        ``request_offsets`` / ``request_num_steps`` are the caller's pinned host
+        vocab_size]``; request ``r`` owns ``request_num_steps[r] * request_num_beams[r]``
+        consecutive rows starting at ``request_offsets[r]``, in beam-major / step-minor
+        order. ``request_offsets`` / ``request_num_steps`` are the caller's pinned host
         tensors and are staged to the device here.
 
+        Beam search changes only where the counts come from, never where the penalty is
+        applied: every row is rewritten here, in place, on raw logits and before
+        temperature -- the same position in the pipeline as the single-beam path, so both
+        keep the ordering ``bias -> penalty -> bans -> temperature -> sampling``. What
+        beam search adds is a per-beam counts row, re-parented each step, and a row ->
+        (slot, beam) mapping so each beam is penalized against its own history.
+
+        With ``max_beam_width == 1`` the pending-token fold stays fused into the same
+        graph, so the whole step is one kernel.
+
         Args:
+            request_num_beams / predecessor_beams: required when ``max_beam_width > 1``.
+                ``request_num_beams`` is the row-layout width the caller packed ``logits``
+                at (the static admission width, not the per-iteration one), so that the
+                row -> beam mapping matches ``request_offsets``. ``predecessor_beams``
+                must still hold the *previous* step's parent map, i.e. this must run
+                before the step's beam sampling overwrites it.
             is_draft_batch: draft batches share this sampler but draw ``py_seq_slot``
                 from a separate numbering space that collides with target slots, so
                 penalizing them would read/write an unrelated target request's
@@ -442,9 +537,39 @@ def apply(
         for request, state in active_requests:
             self._initialize_workspace(request, state, logits.size(-1))
 
-        request_offsets_cuda, request_num_steps_cuda = store.stage_request_metadata(
-            request_offsets, request_num_steps
+        if self._max_beam_width > 1:
+            assert request_num_beams is not None and predecessor_beams is not None, (
+                "beam search requires request_num_beams and predecessor_beams"
+            )
+            num_beams_host = request_num_beams
+        else:
+            # Left None so the packed pass specializes to its single-beam graph.
+            num_beams_host = None
+
+        # Staged ahead of both consumers, since the views last only until the next call.
+        request_offsets_cuda, request_num_steps_cuda, num_beams_cuda = store.stage_request_metadata(
+            request_offsets, request_num_steps, num_beams_host
         )
+
+        if self._max_beam_width > 1:
+            # Both were established by the branch above -- a beam engine always stages a
+            # beam width -- but the narrowing does not survive stage_request_metadata, so
+            # restate it for the type checker.
+            assert predecessor_beams is not None and num_beams_cuda is not None
+            # Re-parent and fold up front. On a single-beam engine the fold stays fused
+            # into the packed graph below; here it cannot, because re-parenting has to
+            # happen first and it is not expressible inside that graph.
+            Fusions.update_beam_occurrence_counts(
+                counts_cuda,
+                store.active_cuda,
+                store.has_previous_token_cuda,
+                store.beam_slot_cuda,
+                new_tokens,
+                predecessor_beams,
+                seq_slots,
+                num_beams_cuda,
+                self._max_beam_width,
+            )
         Fusions.apply_batched_occurrence_penalties(
             logits,
             counts_cuda,
@@ -458,18 +583,30 @@ def apply(
             store.repetition_cuda,
             store.presence_cuda,
             store.frequency_cuda,
+            # None / 1 / True on a single-beam engine: no beam axis in the row mapping and
+            # the fold stays fused in.
+            num_beams_cuda,
+            self._max_beam_width,
+            self._max_beam_width == 1,
         )
-        # Arm has_previous_token for the slots this call penalized (active, num_steps > 0)
-        # so the next apply folds their sampled new_tokens. Done here rather than in the
-        # compiled op because the op's fold reads the flag for every request row; flipping
-        # it in the same graph would make the result depend on execution order within the
-        # kernel.
-        #
-        # The scan is kept on the host deliberately. The same thing can be expressed on
-        # device as active_cuda[seq_slots] & (num_steps > 0), avoiding this loop and the
-        # H2D, but that costs several extra kernel launches and measured 5-7us slower for
-        # batches up to 32 and no better at 64-256: the loop overlaps with the model
-        # forward, the launches do not.
+        self._arm_pending_tokens(requests, request_num_steps)
+
+    def _arm_pending_tokens(
+        self, requests: list[LlmRequest], request_num_steps: torch.Tensor
+    ) -> None:
+        """Arm has_previous_token for the slots this step advanced (active, num_steps > 0).
+
+        The next call then folds their sampled ``new_tokens``. Done on the host rather than
+        in the compiled op because the fold reads the flag for every request row; flipping
+        it in the same graph would make the result depend on execution order within the
+        kernel.
+
+        The scan is kept on the host deliberately. The same thing can be expressed on
+        device as active_cuda[seq_slots] & (num_steps > 0), avoiding this loop and the
+        H2D, but that costs several extra kernel launches and measured 5-7us slower for
+        batches up to 32 and no better at 64-256: the loop overlaps with the model
+        forward, the launches do not.
+        """
         pending_token_slots: list[int] = []
         for request, num_steps in zip(requests, request_num_steps.tolist()):
             slot = request.py_seq_slot
@@ -478,6 +615,6 @@ def apply(
             if self._slots[slot] is not None and num_steps > 0:
                 pending_token_slots.append(slot)
         if pending_token_slots:
-            store.has_previous_token_cuda.index_fill_(
+            self.store.has_previous_token_cuda.index_fill_(
                 0, self._to_device(pending_token_slots, torch.int64), True
             )
diff --git a/tensorrt_llm/_torch/pyexecutor/sampler/sampler.py b/tensorrt_llm/_torch/pyexecutor/sampler/sampler.py
index 2597174fb6d8..3a91dd3fcc7b 100644
--- a/tensorrt_llm/_torch/pyexecutor/sampler/sampler.py
+++ b/tensorrt_llm/_torch/pyexecutor/sampler/sampler.py
@@ -1523,6 +1523,7 @@ def __init__(self, args: Args):
             )
             self._penalty_handler = PenaltyHandler(
                 max_num_sequences=self.max_num_sequences,
+                max_beam_width=self.max_beam_width,
                 device="cuda",
             )
 
@@ -1967,11 +1968,12 @@ def _collect_new_requests_for_setup(
 
     @override
     def validate_request(self, request: LlmRequest) -> None:
-        # Reject unsupported top-p-decay and penalty combinations at admission, so
-        # only the offending request fails (raising later, inside setup_sampler_step
-        # or sampling, would abort the whole executor step).
+        # Reject unsupported top-p-decay combinations at admission, so only the offending
+        # request fails (raising later, inside setup_sampler_step or sampling, would abort
+        # the whole executor step). Occurrence penalties have no unsupported combination
+        # left: beam search is supported, and beam search with speculative decoding is
+        # rejected for the whole sampler in __init__.
         self._top_p_decay.validate_request(request)
-        self._penalty_handler.validate_request(request)
         if self._use_beam_search:
             if request.py_return_log_probs:
                 if request.py_num_logprobs > 1:
@@ -3814,7 +3816,10 @@ def _process_requests(
         )
 
         # Apply repetition/presence/frequency penalties in place, before the greedy fast
-        # path, so both greedy and grouped-sampling logits are penalized.
+        # path, so both greedy and grouped-sampling logits are penalized. With beam search
+        # this also re-parents the per-beam counts, which reads the predecessor map the
+        # step's sampling is about to overwrite -- so it has to stay ahead of sampling.
+        beam_search_store = self.store.beam_search_store
         self._penalty_handler.apply(
             logits_cuda,
             sampling_requests,
@@ -3822,6 +3827,10 @@ def _process_requests(
             seq_slots=seq_slots_cuda,
             request_offsets=sampling_requests_metadata.req_offsets,
             request_num_steps=sampling_requests_metadata.req_num_steps,
+            request_num_beams=sampling_requests_metadata.req_num_beams,
+            predecessor_beams=(
+                beam_search_store.predecessor_beams if beam_search_store is not None else None
+            ),
             # _is_draft_batch reads requests[0]; an empty batch has no penalties to apply
             # anyway, so short-circuit rather than index into it.
             is_draft_batch=bool(sampling_requests) and self._is_draft_batch(sampling_requests),
diff --git a/tensorrt_llm/_torch/pyexecutor/sampler/sampler_strategy.py b/tensorrt_llm/_torch/pyexecutor/sampler/sampler_strategy.py
index fc04afbb51e4..51b69c1d4c24 100644
--- a/tensorrt_llm/_torch/pyexecutor/sampler/sampler_strategy.py
+++ b/tensorrt_llm/_torch/pyexecutor/sampler/sampler_strategy.py
@@ -351,7 +351,7 @@ def sample(
                 logits,
                 beam_width_in=cast(int, beam_width_in),
                 beam_width_out=cast(int, beam_width_out),
-                row_stride=cast(int, row_stride),
+                row_stride=row_stride,
                 beam_search_args=group_metadata,
                 temperature=cast(float, temperature),
                 early_stopping=cast(int, early_stopping),
@@ -1341,7 +1341,7 @@ def sample_grouped_strategies(
             # variable-beam-width step; the op slices down to the live beams.
             rows_per_request = beam_width_in
             if strategies and strategies[0][0] == "beam_search":
-                rows_per_request = cast(BeamSearch, strategies[0]).row_stride
+                rows_per_request = strategies[0].row_stride
             assert logits.size(0) == rows_per_request * len(strategies)
         else:
             assert group_logit_indices.size(0) == beam_width_in * len(strategies)
diff --git a/tests/unittest/_torch/sampler/test_penalties.py b/tests/unittest/_torch/sampler/test_penalties.py
index 6b9826425ddd..e65766577010 100644
--- a/tests/unittest/_torch/sampler/test_penalties.py
+++ b/tests/unittest/_torch/sampler/test_penalties.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from dataclasses import dataclass
 from types import SimpleNamespace
 
 import pytest
@@ -380,6 +381,7 @@ def test_handler_tracks_overlap_and_commits_speculative_tail() -> None:
     slot = 2
     handler = PenaltyHandler(
         max_num_sequences=3,
+        max_beam_width=1,
         device="cuda",
     )
     history = [3]
@@ -458,6 +460,7 @@ def test_regular_handler_slot_reuse_does_not_leak_penalties() -> None:
     vocab = 16
     handler = PenaltyHandler(
         max_num_sequences=1,
+        max_beam_width=1,
         device="cuda",
     )
     new_tokens = torch.zeros(1, 1, 1, dtype=torch.int32, device="cuda")
@@ -486,30 +489,304 @@ def test_regular_handler_slot_reuse_does_not_leak_penalties() -> None:
     torch.testing.assert_close(logits, expected, rtol=1e-4, atol=1e-4)
 
 
-def test_handler_ignores_occurrence_penalties_with_beam_search() -> None:
-    """Beam-search requests never become penalty-active.
+BEAM_PAD = -1
+BEAM_VOCAB = 11
+BEAM_SLOTS = 5
+BEAM_WIDTH = 4
 
-    ``PenaltyHandler.validate_request`` rejects this combination at admission, so the
-    handler should only ever see beam_width == 1 requests. It stays defensive anyway:
-    a beam-search request leaves its slot inactive, and ``apply`` is then a no-op.
+
+@dataclass(frozen=True)
+class _BeamStep:
+    """One decoding step of a scenario; every field carries one row per slot.
+
+    Rows are in ascending slot order, matching the ``seq_slots`` handed to the op.
     """
-    vocab = 16
-    handler = PenaltyHandler(max_num_sequences=1, device="cuda")
-    new_tokens = torch.zeros(1, 1, 1, dtype=torch.int32, device="cuda")
 
-    request = _make_handler_request(slot=0, tokens=[3, 3], beam_width=2)
-    _admit(handler, request, 0)
+    predecessor_beams: list[list[int]]
+    """Parent beam of each beam, as the previous step's beam search would have written."""
+    sampled_tokens: list[list[int]]
+    """Token each beam sampled last step; ``BEAM_PAD`` where the parent had finished."""
+    armed: list[bool]
+    """Whether the slot sampled last step -- the ``has_previous_token`` latch."""
+    num_beams: list[int]
+    """Row-layout beam width; beams at or past it are re-parented but never folded.
+
+    The sampler passes the static admission width here, so in production this only
+    drops the beams of a narrower request sharing the engine; a beam left behind by a
+    growing ``beam_width_array`` is dropped by its ``BEAM_PAD`` instead. The op's
+    contract is the same either way, so the scenarios exercise both gates."""
+
+
+@dataclass(frozen=True)
+class _BeamCountScenario:
+    name: str
+    prompts: dict[int, list[int]]
+    """Prompt tokens per slot, seeded onto every beam of that slot."""
+    steps: list[_BeamStep]
+
+
+_BEAM_COUNT_SCENARIOS = [
+    # Every beam continues itself, so the histories just diverge.
+    _BeamCountScenario(
+        name="divergence",
+        prompts={1: [2, 2, 5], 3: [7]},
+        steps=[
+            _BeamStep(
+                predecessor_beams=[[0, 1, 2, 3], [0, 1, 2, 3]],
+                sampled_tokens=[[1, 2, 3, 4], [5, 6, 7, 8]],
+                armed=[True, True],
+                num_beams=[4, 4],
+            ),
+            _BeamStep(
+                predecessor_beams=[[0, 1, 2, 3], [0, 1, 2, 3]],
+                sampled_tokens=[[2, 2, 2, 2], [7, 7, 7, 7]],
+                armed=[True, True],
+                num_beams=[4, 4],
+            ),
+        ],
+    ),
+    # Several beams share one parent and another parent is dropped entirely -- the
+    # case a per-slot count row cannot represent.
+    _BeamCountScenario(
+        name="reconvergence",
+        prompts={0: [3]},
+        steps=[
+            _BeamStep([[0, 0, 0, 0]], [[1, 2, 3, 4]], [True], [4]),
+            _BeamStep([[2, 2, 0, 2]], [[5, 6, 7, 8]], [True], [4]),
+            _BeamStep([[1, 1, 1, 1]], [[9, 9, 9, 9]], [True], [4]),
+        ],
+    ),
+    # Beam width grows 1 -> BEAM_WIDTH: the context step is unarmed, then the first
+    # generation step re-parents every beam onto beam 0 and folds its own token.
+    _BeamCountScenario(
+        name="vbws-growth",
+        prompts={2: [4, 4]},
+        steps=[
+            _BeamStep([[0, 0, 0, 0]], [[6, BEAM_PAD, BEAM_PAD, BEAM_PAD]], [False], [1]),
+            _BeamStep([[0, 0, 0, 0]], [[6, 7, 8, 9]], [True], [4]),
+            _BeamStep([[0, 1, 2, 3]], [[1, 1, 1, 1]], [True], [4]),
+        ],
+    ),
+    # Intermediate growth 2 -> 4: the previous step's map holds two valid entries and two
+    # stale ones in the same slot, and the new beams inherit from beam 1, whose history
+    # differs from beam 0's -- so a stale row that was not re-parented is detectable.
+    _BeamCountScenario(
+        name="vbws-growth-partial",
+        prompts={0: [3]},
+        steps=[
+            _BeamStep([[0, 0, 0, 0]], [[1, 2, BEAM_PAD, BEAM_PAD]], [True], [2]),
+            _BeamStep([[0, 1, 1, 1]], [[5, 6, 7, 8]], [True], [4]),
+            _BeamStep([[0, 1, 2, 3]], [[9, 9, 9, 9]], [True], [4]),
+        ],
+    ),
+    # A beam whose predecessor already finished emits BEAM_SEARCH_PAD_TOKEN.
+    _BeamCountScenario(
+        name="pad-token",
+        prompts={4: [0]},
+        steps=[
+            _BeamStep([[0, 0, 0, 0]], [[1, 2, BEAM_PAD, BEAM_PAD]], [True], [4]),
+            _BeamStep([[0, 1, 2, 3]], [[BEAM_PAD, 3, BEAM_PAD, 4]], [True], [4]),
+        ],
+    ),
+    # A slot scheduled without sampling keeps a stale predecessor map; the
+    # has_previous_token latch must make the whole step a no-op.
+    _BeamCountScenario(
+        name="unarmed-identity",
+        prompts={1: [8, 8]},
+        steps=[
+            _BeamStep([[0, 0, 0, 0]], [[1, 2, 3, 4]], [True], [4]),
+            _BeamStep([[3, 3, 3, 3]], [[5, 5, 5, 5]], [False], [4]),
+            _BeamStep([[0, 1, 2, 3]], [[6, 6, 6, 6]], [True], [4]),
+        ],
+    ),
+    # Beams at or past the request's layout width are re-parented but never folded --
+    # a narrower request sharing a wider engine.
+    _BeamCountScenario(
+        name="narrow-request",
+        prompts={0: [1]},
+        steps=[
+            _BeamStep([[0, 0, 0, 0]], [[2, 3, 4, 5]], [True], [4]),
+            _BeamStep([[1, 3, 0, 0]], [[6, 7, 0, 0]], [True], [2]),
+        ],
+    ),
+    # Multimodal placeholder ids land outside the vocab and must be dropped.
+    _BeamCountScenario(
+        name="out-of-range-token",
+        prompts={3: [2]},
+        steps=[_BeamStep([[0, 0, 0, 0]], [[BEAM_VOCAB, BEAM_VOCAB + 7, 1, -5]], [True], [4])],
+    ),
+]
+
+
+def _beam_counts_reference(scenario: _BeamCountScenario) -> torch.Tensor:
+    """Host replay of the per-beam occurrence counts.
+
+    Mirrors the op's contract: on an armed slot every beam first takes over the
+    history of ``predecessor_beams[beam]``, then the beams below the layout width
+    append the token they sampled. Unarmed slots are left alone.
+    """
+    slots = sorted(scenario.prompts)
+    history = {
+        (slot, beam): list(prompt)
+        for slot, prompt in scenario.prompts.items()
+        for beam in range(BEAM_WIDTH)
+    }
+    for step in scenario.steps:
+        advanced = dict(history)
+        for index, slot in enumerate(slots):
+            if not step.armed[index]:
+                continue
+            for beam in range(BEAM_WIDTH):
+                inherited = list(history[(slot, step.predecessor_beams[index][beam])])
+                token = step.sampled_tokens[index][beam]
+                if beam < step.num_beams[index] and 0 <= token < BEAM_VOCAB:
+                    inherited.append(token)
+                advanced[(slot, beam)] = inherited
+        history = advanced
+
+    expected = torch.zeros((BEAM_SLOTS * BEAM_WIDTH, BEAM_VOCAB), dtype=torch.int32, device="cuda")
+    for (slot, beam), tokens in history.items():
+        for token in tokens:
+            expected[slot * BEAM_WIDTH + beam, token] += 1
+    return expected
 
-    logits = torch.linspace(-2.0, 2.0, steps=vocab, device="cuda").view(1, vocab)
-    original = logits.clone()
-    _apply_handler(handler, request, logits, 1, new_tokens)
 
-    assert not bool(handler.store.active_cuda[0].item())
-    torch.testing.assert_close(logits, original, rtol=0, atol=0)
+@pytest.mark.parametrize(
+    "scenario", _BEAM_COUNT_SCENARIOS, ids=[s.name for s in _BEAM_COUNT_SCENARIOS]
+)
+def test_beam_occurrence_counts_follow_each_beam_history(scenario: _BeamCountScenario) -> None:
+    """``update_beam_occurrence_counts`` keeps one true history per beam.
+
+    Each beam inherits its parent's counts and appends its own token, the torch
+    counterpart of the C++ ``batchApplyPenalty`` workspace copy along ``parentIds``.
+    """
+    counts = torch.zeros((BEAM_SLOTS * BEAM_WIDTH, BEAM_VOCAB), dtype=torch.int32, device="cuda")
+    active = torch.zeros(BEAM_SLOTS, dtype=torch.bool, device="cuda")
+    has_previous_token = torch.zeros(BEAM_SLOTS, dtype=torch.bool, device="cuda")
+    predecessor_beams = torch.zeros((BEAM_SLOTS, BEAM_WIDTH), dtype=torch.int32, device="cuda")
+    new_tokens = torch.zeros(1, BEAM_SLOTS, BEAM_WIDTH, dtype=torch.int32, device="cuda")
+    slots = sorted(scenario.prompts)
+    seq_slots = torch.tensor(slots, dtype=torch.int64, device="cuda")
+
+    # Seed every beam from the prompt, as PenaltyHandler._initialize_workspace does.
+    for slot, prompt in scenario.prompts.items():
+        active[slot] = True
+        for beam in range(BEAM_WIDTH):
+            for token in prompt:
+                counts[slot * BEAM_WIDTH + beam, token] += 1
+
+    for step in scenario.steps:
+        for index, slot in enumerate(slots):
+            predecessor_beams[slot] = torch.tensor(
+                step.predecessor_beams[index], dtype=torch.int32, device="cuda"
+            )
+            new_tokens[0, slot] = torch.tensor(
+                step.sampled_tokens[index], dtype=torch.int32, device="cuda"
+            )
+            has_previous_token[slot] = step.armed[index]
+        Fusions.update_beam_occurrence_counts(
+            counts,
+            active,
+            has_previous_token,
+            torch.ones(BEAM_SLOTS, dtype=torch.bool, device="cuda"),
+            new_tokens,
+            predecessor_beams,
+            seq_slots,
+            torch.tensor(step.num_beams, dtype=torch.int32, device="cuda"),
+            BEAM_WIDTH,
+        )
+
+    torch.testing.assert_close(counts, _beam_counts_reference(scenario), rtol=0, atol=0)
+
+
+def test_single_beam_slot_sharing_a_beam_engine_is_routed_correctly() -> None:
+    """A width-1 request on a beam engine must not be treated as a beam request.
+
+    Beam width is per request, so once TRTLLM-14792 lifts the equal-width restriction a
+    batch can mix the two. The single-beam slot must never be re-parented -- nothing
+    writes its ``predecessor_beams`` row, so believing it would corrupt its history --
+    while both kinds are penalized by the same packed pass, each against its own row.
+    """
+    max_beam_width, vocab, num_slots = 4, 32, 2
+    beam_slot, plain_slot = 0, 1
+
+    counts = torch.zeros((num_slots * max_beam_width, vocab), dtype=torch.int32, device="cuda")
+    active = torch.ones(num_slots, dtype=torch.bool, device="cuda")
+    armed = torch.ones(num_slots, dtype=torch.bool, device="cuda")
+    is_beam = torch.tensor([True, False], dtype=torch.bool, device="cuda")
+    seq_slots = torch.tensor([beam_slot, plain_slot], dtype=torch.int64, device="cuda")
+
+    # Give the plain slot a hostile predecessor map: if it were believed, beam 0 would
+    # inherit beam 3's counts.
+    predecessor_beams = torch.zeros((num_slots, max_beam_width), dtype=torch.int32, device="cuda")
+    predecessor_beams[plain_slot, :] = 3
+    counts[plain_slot * max_beam_width + 3, 9] = 7  # poison beam 3 of the plain slot
+
+    new_tokens = torch.zeros(1, num_slots, max_beam_width, dtype=torch.int32, device="cuda")
+    new_tokens[0, beam_slot] = torch.tensor([1, 2, 3, 4], dtype=torch.int32, device="cuda")
+    new_tokens[0, plain_slot] = torch.tensor([5, 6, 7, 8], dtype=torch.int32, device="cuda")
 
+    Fusions.update_beam_occurrence_counts(
+        counts,
+        active,
+        armed,
+        is_beam,
+        new_tokens,
+        predecessor_beams,
+        seq_slots,
+        torch.tensor([max_beam_width, 1], dtype=torch.int32, device="cuda"),
+        max_beam_width,
+    )
 
-def test_validate_request_rejects_penalties_with_beam_search() -> None:
-    """The admission-time check that keeps the combination above from arriving."""
-    PenaltyHandler.validate_request(_make_handler_request(slot=0, tokens=[3], beam_width=1))
-    with pytest.raises(ValueError, match="penalties with beam search"):
-        PenaltyHandler.validate_request(_make_handler_request(slot=0, tokens=[3], beam_width=2))
+    plain_base = plain_slot * max_beam_width
+    assert int(counts[plain_base, 9].item()) == 0, "plain slot must not inherit beam 3"
+    assert int(counts[plain_base, 5].item()) == 1, "plain slot folds only its own token"
+    for token in (6, 7, 8):
+        assert int(counts[plain_base, token].item()) == 0, "beams 1..3 must not fold"
+    # The beam slot still behaves as before: every beam folds its own token.
+    for beam, token in enumerate((1, 2, 3, 4)):
+        assert int(counts[beam_slot * max_beam_width + beam, token].item()) == 1
+
+    # The packed pass penalizes both kinds; the plain slot must read its beam-0 row and
+    # not, say, beam 3's poisoned counts.
+    rows = max_beam_width + 1  # beam slot: 4 rows; plain slot: 1
+    logits = torch.linspace(-2.0, 2.0, steps=rows * vocab, device="cuda").view(rows, vocab)
+    original = logits.clone()
+    rep = torch.full((num_slots,), 2.0, device="cuda")
+    zero = torch.zeros(num_slots, device="cuda")
+    apply_batched_occurrence_penalties(
+        logits,
+        counts,
+        None,
+        active,
+        armed,
+        new_tokens,
+        seq_slots,
+        torch.tensor([0, max_beam_width], dtype=torch.int32, device="cuda"),
+        torch.tensor([1, 1], dtype=torch.int32, device="cuda"),
+        rep,
+        zero,
+        zero,
+        torch.tensor([max_beam_width, 1], dtype=torch.int32, device="cuda"),
+        max_beam_width,
+        False,  # the fold already happened above
+    )
+
+    def penalized(x: torch.Tensor) -> torch.Tensor:
+        return torch.where(x < 0, x * 2.0, x / 2.0)
+
+    plain_row = max_beam_width
+    # Token 5 is the plain slot's only counted token -> repetition branch.
+    torch.testing.assert_close(
+        logits[plain_row, 5], penalized(original[plain_row, 5]), rtol=1e-5, atol=1e-5
+    )
+    # Token 9 is only in the poisoned beam-3 row, which the plain slot must not read.
+    torch.testing.assert_close(logits[plain_row, 9], original[plain_row, 9], rtol=0, atol=0)
+    # Each beam row of the beam slot is penalized against its own token.
+    for beam, token in enumerate((1, 2, 3, 4)):
+        torch.testing.assert_close(
+            logits[beam, token], penalized(original[beam, token]), rtol=1e-5, atol=1e-5
+        )
+        other = 1 + ((beam + 1) % 4)  # a token belonging to a different beam
+        if other != token:
+            torch.testing.assert_close(logits[beam, other], original[beam, other], rtol=0, atol=0)
diff --git a/tests/unittest/_torch/sampler/test_penalties_e2e.py b/tests/unittest/_torch/sampler/test_penalties_e2e.py
index 332f8a675c14..553aa51717c4 100644
--- a/tests/unittest/_torch/sampler/test_penalties_e2e.py
+++ b/tests/unittest/_torch/sampler/test_penalties_e2e.py
@@ -12,14 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from collections import Counter
 from dataclasses import dataclass
 from pathlib import Path
 
 import pytest
 import torch
+from test_beam_search_util import DummyConfigLoader, DummyWeightLoader
 from utils.llm_data import llm_models_root
 
 from tensorrt_llm import LLM, SamplingParams
+from tensorrt_llm._torch.models.checkpoints import HfCheckpointLoader
 from tensorrt_llm.executor.result import CompletionOutput, GenerationResult
 from tensorrt_llm.llmapi import CudaGraphConfig, NGramDecodingConfig
 from tensorrt_llm.llmapi import KvCacheConfig as TRT_KvCacheConfig
@@ -327,3 +330,116 @@ def test_torch_sampler_speculative_penalty_e2e(model_path: Path) -> None:
     )
     for completion in speculative_outputs[0].outputs:
         _assert_completion_penalty_logprobs(case, completion, speculative_prompt_token_ids)
+
+
+# --- Beam search + occurrence penalties ------------------------------------------
+# DummyModel (test_beam_search_util) has analytically predictable logits that depend
+# only on the current input token, so the whole beam search can be replayed on the
+# host and compared token for token. That is what makes this an actual correctness
+# check of the wiring -- the per-request num_beams and the predecessor map the
+# re-parenting reads -- rather than a smoke test.
+
+_E2E_VOCAB = 1000  # DummyConfig.vocab_size
+_E2E_BEAM_WIDTH = 2
+_E2E_MAX_TOKENS = 4
+# Ends with 3, so the candidates are 3/4/5/6 scoring 0.3/0.6/0.9/1.2. Token 6 is in
+# the prompt, so a repetition penalty demotes the otherwise-winning candidate and the
+# selected beams change outright -- see the != assertion at the end of the test.
+_E2E_PROMPT = [6, 1, 2, 3]
+_E2E_REPETITION_PENALTY = 3.0
+
+
+def _dummy_model_logits(token: int) -> torch.Tensor:
+    """DummyModel.forward's logits for one input token (see test_beam_search_util)."""
+    logits = torch.zeros(_E2E_VOCAB, dtype=torch.float32)
+    for offset in range(4):
+        logits[(token + offset) % _E2E_VOCAB] += 0.1 * (offset + 1) * token
+    return logits
+
+
+def _replay_beam_search(repetition_penalty: float) -> list[list[int]]:
+    """Independent host replay of beam search under a repetition penalty.
+
+    Mirrors the pipeline order the sampler uses: penalize the raw logits, then
+    log_softmax, then add the running cumulative log-prob, then take the top
+    ``beam_width`` over the flattened (beam, vocab) scores. Each surviving beam
+    inherits its parent's occurrence counts and appends its own token, which is
+    exactly what ``update_beam_occurrence_counts`` does on the device.
+    """
+    # (generated tokens, cumulative log-prob, occurrence counts). The prompt seeds the
+    # counts, and the first step expands the single context beam into beam_width beams.
+    beams: list[tuple[list[int], float, Counter]] = [([], 0.0, Counter(_E2E_PROMPT))]
+    last_tokens = [_E2E_PROMPT[-1]]
+
+    for _ in range(_E2E_MAX_TOKENS):
+        scores = []
+        for (_, cum_log_prob, counts), last_token in zip(beams, last_tokens):
+            logits = _dummy_model_logits(last_token)
+            for token, count in counts.items():
+                if count > 0 and token < _E2E_VOCAB:
+                    value = logits[token].item()
+                    logits[token] = (
+                        value * repetition_penalty if value < 0 else value / repetition_penalty
+                    )
+            scores.append(torch.log_softmax(logits, dim=-1) + cum_log_prob)
+
+        flat = torch.stack(scores).reshape(-1)
+        top = torch.topk(flat, _E2E_BEAM_WIDTH)
+        advanced, advanced_last = [], []
+        for rank in range(_E2E_BEAM_WIDTH):
+            parent, token = divmod(int(top.indices[rank].item()), _E2E_VOCAB)
+            parent_tokens, _, parent_counts = beams[parent]
+            counts = parent_counts.copy()
+            counts[token] += 1
+            advanced.append((parent_tokens + [token], float(top.values[rank].item()), counts))
+            advanced_last.append(token)
+        beams, last_tokens = advanced, advanced_last
+
+    return [tokens for tokens, _, _ in beams]
+
+
+@pytest.mark.parametrize("overlap", [False, True], ids=["no_overlap", "overlap"])
+def test_beam_search_penalties_e2e(overlap: bool) -> None:
+    """The full pipeline must reproduce an independent replay of the algorithm.
+
+    This is the only coverage of ``TorchSampler`` actually handing the penalty handler
+    the per-request beam widths and the beam store's predecessor map, and of that
+    happening before the step's sampling overwrites the latter.
+    """
+    llm = LLM(
+        model=Path("dummy_path"),
+        checkpoint_loader=HfCheckpointLoader(
+            weight_loader=DummyWeightLoader(),
+            config_loader=DummyConfigLoader(),
+        ),
+        sampler_type="TorchSampler",
+        max_batch_size=_E2E_BEAM_WIDTH,
+        kv_cache_config=TRT_KvCacheConfig(max_tokens=10000),
+        max_seq_len=32,
+        max_beam_width=_E2E_BEAM_WIDTH,
+        disable_overlap_scheduler=not overlap,
+        cuda_graph_config=None,
+    )
+    with llm:
+        sampling_params = SamplingParams(
+            max_tokens=_E2E_MAX_TOKENS,
+            use_beam_search=True,
+            best_of=_E2E_BEAM_WIDTH,
+            n=_E2E_BEAM_WIDTH,
+            temperature=1.0,
+            repetition_penalty=_E2E_REPETITION_PENALTY,
+            # The dummy checkpoint has no tokenizer to derive an end id from, and no
+            # token must terminate generation early: the replay assumes all max_tokens.
+            end_id=-1,
+        )
+        outputs = llm.generate([_E2E_PROMPT], sampling_params=sampling_params)
+
+    got = [list(beam.token_ids) for beam in outputs[0].outputs]
+    expected = _replay_beam_search(_E2E_REPETITION_PENALTY)
+    assert got == expected, f"penalized beams {got} != replay {expected}"
+
+    # The penalty must actually have changed the outcome, otherwise the comparison
+    # above would pass just as well with the penalty never applied.
+    assert expected != _replay_beam_search(1.0), (
+        "test setup no longer distinguishes penalized from unpenalized beams"
+    )