diff --git a/python/sglang/srt/model_executor/breakable_cuda_graph_runner.py b/python/sglang/srt/model_executor/breakable_cuda_graph_runner.py
index 7d2a495c9..6dada9c31 100644
--- a/python/sglang/srt/model_executor/breakable_cuda_graph_runner.py
+++ b/python/sglang/srt/model_executor/breakable_cuda_graph_runner.py
@@ -169,6 +169,9 @@ class BreakableCudaGraphRunner:
 
     def _init_buffers(self, model_runner):
         """Initialize input buffers."""
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_prefill_registry,
+        )
         from sglang.srt.model_executor.piecewise_cuda_graph_runner import (
             PrefillInputBuffers,
         )
@@ -215,6 +218,21 @@ class BreakableCudaGraphRunner:
         )
         self.buffers.share_buffers()
 
+        # Token-axis FB-shared slot registry adopting the PrefillInputBuffers
+        # storage. Breakable has no mamba track and bs is not padded here, so
+        # there are no bs-axis slots (max_bs is unused).
+        self.buffer_registry = build_prefill_registry(
+            device=self.device,
+            max_bs=1,
+            max_num_token=self.max_num_tokens,
+            cache_loc_dtype=torch.int64 if not is_npu() else torch.int32,
+            is_multimodal=self.is_multimodal,
+            hidden_size=model_runner.model_config.hidden_size,
+            embed_dtype=model_runner.dtype,
+            enable_mamba_track=False,
+            source=self.buffers,
+        )
+
     @torch.no_grad()
     def _run_forward(self, forward_batch, num_tokens):
         """Run layer-stack forward with proper context.
@@ -271,8 +289,12 @@ class BreakableCudaGraphRunner:
                 hidden_states=self.static_draft_hidden_states[:num_tokens],
             )
 
-        buffers = self.buffers
+        registry = self.buffer_registry
         bs = 1
+
+        def _slot(name):
+            return registry.get_slot(name).slice_for(bs, num_tokens)
+
         with torch.device(self.device):
             seq_lens = torch.full((bs,), num_tokens, dtype=torch.int64)
             extend_seq_lens = torch.full((bs,), num_tokens, dtype=torch.int64)
@@ -284,16 +306,16 @@ class BreakableCudaGraphRunner:
         return ForwardBatch(
             forward_mode=ForwardMode.EXTEND,
             batch_size=bs,
-            input_ids=buffers.input_ids[:num_tokens],
+            input_ids=_slot("input_ids"),
             input_embeds=(
-                buffers.input_embeds[:num_tokens] if self.is_multimodal else None
+                _slot("input_embeds") if registry.has_slot("input_embeds") else None
             ),
             req_pool_indices=req_pool_indices,
             seq_lens=seq_lens,
             next_token_logits_buffer=None,
             orig_seq_lens=orig_seq_lens,
             seq_lens_cpu=torch.tensor([num_tokens], device="cpu"),
-            out_cache_loc=buffers.out_cache_loc[:num_tokens],
+            out_cache_loc=_slot("out_cache_loc"),
             seq_lens_sum=num_tokens,
             mamba_track_indices=None,
             mamba_track_mask=None,
@@ -307,13 +329,15 @@ class BreakableCudaGraphRunner:
             extend_prefix_lens_cpu=torch.tensor([0], device="cpu"),
             extend_seq_lens_cpu=torch.tensor([num_tokens], device="cpu"),
             extend_logprob_start_lens_cpu=torch.tensor([num_tokens], device="cpu"),
-            positions=buffers.positions[:num_tokens],
+            positions=_slot("positions"),
             global_num_tokens_gpu=None,
             global_num_tokens_for_logprob_gpu=None,
             dp_padding_mode=DpPaddingMode.get_default_mode_in_cuda_graph(),
             global_dp_buffer_len=None,
             mrope_positions=(
-                buffers.mrope_positions[:, :num_tokens] if self.is_multimodal else None
+                _slot("mrope_positions")
+                if registry.has_slot("mrope_positions")
+                else None
             ),
             spec_algorithm=None,
             spec_info=spec_info,
@@ -446,9 +470,9 @@ class BreakableCudaGraphRunner:
             if self.use_input_embeds:
                 if ie is None:
                     raise ValueError("BCG replay expects input_embeds but got None")
-                self.buffers.input_embeds[:static_num_tokens].copy_(
-                    ie[:static_num_tokens]
-                )
+                self.buffer_registry.get_slot("input_embeds").slice_for(
+                    1, static_num_tokens
+                ).copy_(ie[:static_num_tokens])
             else:
                 if ie is not None:
                     raise ValueError(
diff --git a/python/sglang/srt/model_executor/cuda_graph_buffer_registry.py b/python/sglang/srt/model_executor/cuda_graph_buffer_registry.py
new file mode 100644
index 000000000..90e9fadc4
--- /dev/null
+++ b/python/sglang/srt/model_executor/cuda_graph_buffer_registry.py
@@ -0,0 +1,848 @@
+"""FB-shared slot registry for the CUDA graph forward paths.
+
+``CudaGraphBufferRegistry`` is the ForwardBatch → graph-resident buffer mirror
+used by capture / replay. It replaces the per-runner ``DecodeInputBuffers`` /
+``PrefillInputBuffers`` dataclasses and their hand-written
+``populate_from_forward_batch`` methods with a single ``GraphSlot``-driven
+registry.
+
+Backend-private buffers (kernel workspaces, derived page tables, etc.) stay
+on ``AttentionBackend.cuda_graph_*`` — the registry only owns FB-shared
+slots (FB attribute name maps 1:1 to slot name).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple
+
+import torch
+
+from sglang.srt.model_executor.input_buffers import share_input_buffer
+
+if TYPE_CHECKING:
+    from sglang.srt.model_executor.forward_batch_info import ForwardBatch
+
+
+_has_foreach_copy = hasattr(torch, "_foreach_copy_")
+
+
+def _grouped_foreach_copy_(dsts: List[torch.Tensor], srcs: List[torch.Tensor]) -> None:
+    """Call torch._foreach_copy_ grouped by (dst_dtype, src_dtype) pairs
+    (a single foreach call requires a uniform dtype pair)."""
+
+    def _foreach_copy(
+        group_dsts: List[torch.Tensor], group_srcs: List[torch.Tensor]
+    ) -> None:
+        if _has_foreach_copy:
+            torch._foreach_copy_(group_dsts, group_srcs)
+        else:
+            for dst, src in zip(group_dsts, group_srcs):
+                dst.copy_(src)
+
+    groups: Dict[Tuple[torch.dtype, torch.dtype], Tuple[List, List]] = {}
+    for dst, src in zip(dsts, srcs):
+        key = (dst.dtype, src.dtype)
+        if key not in groups:
+            groups[key] = ([], [])
+        groups[key][0].append(dst)
+        groups[key][1].append(src)
+    for group_dsts, group_srcs in groups.values():
+        _foreach_copy(group_dsts, group_srcs)
+
+
+class PaddingPolicy(Enum):
+    """How to handle ``raw_n < padded_n`` for a slot.
+
+    KEEP_PAD      — Leave the padded region as-is (caller proves the
+                    padded tail will not be read).
+    FILL_SENTINEL — Reset the padded region to ``slot.pad_value`` before
+                    copy (e.g. ``seq_lens`` filled with
+                    ``seq_len_fill_value``).
+    ZERO          — Reset the padded region to ``0`` (e.g.
+                    ``out_cache_loc`` / ``req_pool_indices`` — padded
+                    rows must point at slot 0 so dummy attention reads
+                    land harmlessly).
+    FOREACH_COPY  — Always copy ``raw_n`` from src; padded region is
+                    left as whatever the previous replay (or the init
+                    zeros) wrote. Caller is responsible for proving
+                    safety.
+    FILL_ONCE     — Fill the whole buffer to ``pad_value`` once at alloc;
+                    never reset per iter (e.g. ``encoder_lens`` init to
+                    ``encoder_len_fill_value``, copied head-only with the
+                    tail kept).
+    """
+
+    KEEP_PAD = "keep_pad"
+    FILL_SENTINEL = "fill_sentinel"
+    ZERO = "zero"
+    FOREACH_COPY = "foreach_copy"
+    FILL_ONCE = "fill_once"
+
+
+@dataclass
+class FillContext:
+    """Per-iteration shape context passed to ``GraphSlot.post_fill``.
+
+    Carries both the bs-axis and tokens-axis raw/padded counts so a hook can
+    derive values regardless of its own slot's axis — e.g. the padded token
+    count (``padded_num_tokens`` == padded_bs * num_tokens_per_bs), which the
+    global-num-tokens fill and the local-num-token-non-padded transform need.
+    """
+
+    raw_bs: int
+    padded_bs: int
+    raw_num_tokens: int
+    padded_num_tokens: int
+    # Side inputs that are not ForwardBatch attributes but are needed by a
+    # slot's source_fn — e.g. the pipeline-parallel proxy tensors, which the
+    # replay path receives as a separate argument rather than off the FB.
+    pp_proxy_tensors: Optional[Any] = None
+
+
+@dataclass
+class GraphSlot:
+    """A single FB-mirrored buffer.
+
+    Each slot mirrors one ``ForwardBatch`` attribute. ``name`` MUST match
+    the FB attribute name so ``fill_from`` can ``getattr(fb, name)`` and
+    ``extract_buffer`` can ``setattr`` the view back into a FB replace.
+
+    Fields:
+        name           — the FB attribute name mirrored by this slot.
+        shape_fn       — ``(max_bs, max_num_tokens) -> shape`` callable
+                         used at ``register_slot`` time to allocate the
+                         physical buffer.
+        dtype          — buffer dtype.
+        axis           — ``"bs"`` (slot is sliced ``[:bs]``) or
+                         ``"tokens"`` (sliced ``[:num_tokens]``) or
+                         ``"none"`` (no slicing — full buffer always
+                         exposed; used for scalar buffers and global
+                         counters).
+        device         — buffer device. ``None`` means use registry
+                         default; can be ``"cpu"`` for slots like
+                         ``seq_lens_cpu`` that must live on host.
+        padding_policy — see ``PaddingPolicy``.
+        pad_value      — sentinel for ``FILL_SENTINEL``.
+        enabled        — runtime gate; disabled slots are not allocated
+                         and skipped during fill / extract.
+        copy_from_fb   — when ``True`` (default), ``fill_from`` copies the
+                         same-named FB tensor into the buffer head. Set
+                         ``False`` for computed slots whose value is not a
+                         straight FB copy (e.g. ``global_num_tokens_*``,
+                         filled by a ``post_fill`` instead).
+        post_fill      — optional ``(buffer, forward_batch, FillContext)
+                         -> None`` hook run after the grouped copy. Used for
+                         compute-then-write slots (local-num-token-non-padded
+                         transform, global-num-tokens fill).
+        slice_fn       — optional ``(buffer, padded_n) -> Tensor``
+                         override for slots with non-trivial slicing
+                         (e.g. ``mrope_positions`` shape ``[3, T]`` is
+                         sliced on axis 1 not 0).
+        source_fn      — optional ``(forward_batch, FillContext) -> Tensor |
+                         None`` override for the copy *source*. When set,
+                         ``fill_from`` copies ``source_fn(fb, ctx)`` (instead of
+                         the same-named FB attribute) into
+                         ``buffer[:src.shape[0]]`` — a source-length slice for
+                         structured / side-sourced fields whose data lives on a
+                         nested FB dataclass (``ngram_embedding_info.*``) or an
+                         out-of-band argument (``pp_proxy_tensors``, carried on
+                         ``FillContext``). Returning ``None`` skips the copy for
+                         that iteration. Such slots use dotted names and are
+                         skipped by ``extract_buffer``.
+    """
+
+    name: str
+    shape_fn: Callable[[int, int], Tuple[int, ...]]
+    dtype: torch.dtype
+    axis: str = "tokens"
+    device: Optional[torch.device] = None
+    padding_policy: PaddingPolicy = PaddingPolicy.FOREACH_COPY
+    pad_value: Optional[Any] = None
+    enabled: bool = True
+    copy_from_fb: bool = True
+    post_fill: Optional[
+        Callable[[torch.Tensor, "ForwardBatch", "FillContext"], None]
+    ] = None
+    slice_fn: Optional[Callable[[torch.Tensor, int], torch.Tensor]] = None
+    source_fn: Optional[
+        Callable[["ForwardBatch", "FillContext"], Optional[torch.Tensor]]
+    ] = None
+
+    # runtime
+    buffer: Optional[torch.Tensor] = field(default=None, repr=False)
+
+    def __post_init__(self) -> None:
+        if self.axis not in ("bs", "tokens", "none"):
+            raise ValueError(
+                f"GraphSlot {self.name!r}: axis must be one of "
+                f"'bs'/'tokens'/'none', got {self.axis!r}"
+            )
+
+    def _padded_n(self, padded_bs: int, padded_num_tokens: int) -> int:
+        if self.axis == "bs":
+            return padded_bs
+        if self.axis == "tokens":
+            return padded_num_tokens
+        # axis == "none": no slicing
+        return self.buffer.shape[0] if self.buffer is not None else 0
+
+    def _raw_n(self, raw_bs: int, raw_num_tokens: int) -> int:
+        if self.axis == "bs":
+            return raw_bs
+        if self.axis == "tokens":
+            return raw_num_tokens
+        return self.buffer.shape[0] if self.buffer is not None else 0
+
+    def slice_for(self, padded_bs: int, padded_num_tokens: int) -> torch.Tensor:
+        """Return the ``[:padded_n]`` slice of the buffer consumed by callers.
+
+        This truncates the (full-length) buffer to the active region for the
+        current iteration — it is a slice, not a tensor reshape.
+        """
+        if self.buffer is None:
+            raise RuntimeError(f"GraphSlot {self.name!r}: buffer not allocated")
+        if self.slice_fn is not None:
+            return self.slice_fn(
+                self.buffer, self._padded_n(padded_bs, padded_num_tokens)
+            )
+        if self.axis == "none":
+            return self.buffer
+        return self.buffer[: self._padded_n(padded_bs, padded_num_tokens)]
+
+    def reset_padding(self, raw_n: int, padded_n: int) -> None:
+        """Reset the padded tail according to ``padding_policy``."""
+        if self.buffer is None or raw_n >= padded_n:
+            return
+        if self.padding_policy in (
+            PaddingPolicy.KEEP_PAD,
+            PaddingPolicy.FOREACH_COPY,
+            PaddingPolicy.FILL_ONCE,
+        ):
+            return
+        # slice_fn governs non-trivial layouts (e.g. mrope_positions [3, T]);
+        # the pad region is the same axis the slot exposes via slice_for().
+        if self.slice_fn is not None:
+            # slice_fn returns the [:padded_n] portion already; we need the
+            # tail [raw_n:padded_n]. We rely on slice_fn slicing the same
+            # axis used by slice_for(): take the padded slice first, then index
+            # the tail with the standard slice on axis 0 of the result.
+            padded_slice = self.slice_fn(self.buffer, padded_n)
+            tail = (
+                padded_slice[..., raw_n:padded_n]
+                if padded_slice.dim() > 1
+                else padded_slice[raw_n:padded_n]
+            )
+        else:
+            tail = self.buffer[raw_n:padded_n]
+        if self.padding_policy == PaddingPolicy.FILL_SENTINEL:
+            if self.pad_value is None:
+                raise RuntimeError(
+                    f"GraphSlot {self.name!r}: FILL_SENTINEL requires pad_value"
+                )
+            tail.fill_(self.pad_value)
+        elif self.padding_policy == PaddingPolicy.ZERO:
+            tail.zero_()
+
+
+class CudaGraphBufferRegistry:
+    """FB → graph-resident buffer mirror, shared across eager / capture / replay.
+
+    The registry holds a dict of ``GraphSlot`` instances, each mirroring
+    one ``ForwardBatch`` attribute. Slots are registered up-front (during
+    runner init), allocated at ``register_slot``, then filled per-iter via
+    ``fill_from(fb, ...)`` and consumed via ``extract_buffer(template) ->
+    ForwardBatch``. ``fill_from`` issues plain D2D copies on the caller's
+    current stream; cross-stream correctness (stream handoff) is handled by
+    the runners, not here.
+
+    Backend-private buffers (kernel workspace, derived page tables) are
+    NOT managed here — backends keep them on ``self.cuda_graph_*`` and
+    allocate via ``AttentionBackend.init_cuda_graph_state(...)``.
+
+    Usage::
+
+        registry = CudaGraphBufferRegistry(device=..., max_bs=..., max_num_tokens=...)
+        registry.register_slot(GraphSlot(name="input_ids", ...))
+        registry.register_slot(GraphSlot(name="seq_lens",
+                                         padding_policy=PaddingPolicy.FILL_SENTINEL,
+                                         pad_value=seq_len_fill_value, ...))
+        # per-iter:
+        registry.fill_from(fb, raw_bs=..., padded_bs=..., raw_num_tokens=...,
+                           padded_num_tokens=...)
+        fb_view = registry.extract_buffer(padded_bs=..., padded_num_tokens=...,
+                                          forward_batch_template=fb)
+        attn_backend.init_forward_metadata(fb_view)
+        model.forward(fb_view.input_ids, fb_view.positions, fb_view)
+    """
+
+    def __init__(
+        self,
+        *,
+        device: torch.device,
+        max_bs: int,
+        max_num_tokens: int,
+        share_pool: bool = False,
+    ) -> None:
+        self.device = device
+        self.max_bs = max_bs
+        self.max_num_tokens = max_num_tokens
+        # When True, slot buffers are coalesced by name through the global
+        # ForwardInputBuffers pool, so a registry can share physical storage
+        # (and data_ptr) with the legacy DecodeInputBuffers during migration.
+        self.share_pool = share_pool
+        self._slots: Dict[str, GraphSlot] = {}
+
+    # ---- registration ------------------------------------------------------
+
+    def register_slot(
+        self, slot: GraphSlot, bind: Optional[torch.Tensor] = None
+    ) -> GraphSlot:
+        """Register a slot and allocate (or adopt) its physical buffer.
+
+        If ``bind`` is given, the slot adopts that existing tensor instead of
+        allocating a fresh one (and skips the pool / sentinel init — the bound
+        tensor is assumed already initialized). This lets a registry share
+        storage with the legacy ``DecodeInputBuffers`` by adopting its fields,
+        guaranteeing a stable, identical ``data_ptr`` for capture vs replay.
+
+        Returns the slot for caller convenience. Re-registering an existing
+        name raises.
+        """
+        if slot.name in self._slots:
+            raise ValueError(
+                f"GraphSlot {slot.name!r} already registered; "
+                "use enable()/disable() to gate per-iter."
+            )
+        if not slot.enabled:
+            # Even when disabled, keep the spec so callers can introspect
+            # by name; just don't allocate.
+            self._slots[slot.name] = slot
+            return slot
+        shape = slot.shape_fn(self.max_bs, self.max_num_tokens)
+        device = slot.device if slot.device is not None else self.device
+        if bind is not None:
+            expected = tuple(shape)
+            if tuple(bind.shape) != expected:
+                raise ValueError(
+                    f"bind tensor for slot {slot.name!r} has shape "
+                    f"{tuple(bind.shape)}, expected {expected}."
+                )
+            if bind.dtype != slot.dtype:
+                raise ValueError(
+                    f"bind tensor for slot {slot.name!r} has dtype {bind.dtype}, "
+                    f"expected {slot.dtype}."
+                )
+            slot.buffer = bind
+            self._slots[slot.name] = slot
+            return slot
+        buffer = torch.zeros(shape, dtype=slot.dtype, device=device)
+        if self.share_pool:
+            # Coalesce with any same-named buffer (e.g. the legacy
+            # DecodeInputBuffers field) so capture and replay see one
+            # physical allocation with a stable data_ptr.
+            buffer = share_input_buffer(slot.name, buffer)
+        if (
+            slot.padding_policy
+            in (PaddingPolicy.FILL_SENTINEL, PaddingPolicy.FILL_ONCE)
+            and slot.pad_value is not None
+        ):
+            buffer.fill_(slot.pad_value)
+        slot.buffer = buffer
+        self._slots[slot.name] = slot
+        return slot
+
+    def has_slot(self, name: str) -> bool:
+        return name in self._slots and self._slots[name].enabled
+
+    def get_slot(self, name: str) -> GraphSlot:
+        return self._slots[name]
+
+    def slot_names(self) -> List[str]:
+        return [name for name, s in self._slots.items() if s.enabled]
+
+    # ---- per-iter ----------------------------------------------------------
+
+    def fill_from(
+        self,
+        forward_batch: "ForwardBatch",
+        *,
+        raw_bs: int,
+        padded_bs: int,
+        raw_num_tokens: int,
+        padded_num_tokens: int,
+        pp_proxy_tensors: Optional[Any] = None,
+    ) -> None:
+        """Copy FB → registry buffers.
+
+        Phase 1 — reset the padded tail per slot ``padding_policy``.
+        Phase 2 — grouped D2D copy of all enabled slots from FB (or from a
+                  slot's ``source_fn`` for structured / side-sourced fields).
+        Phase 3 — run ``post_fill`` hooks for slots that need
+                  post-copy transforms.
+
+        ``pp_proxy_tensors`` is the out-of-band pipeline-parallel input; it is
+        not an FB attribute, so it reaches ``source_fn`` slots via
+        ``FillContext.pp_proxy_tensors``.
+
+        Slots whose FB attribute (or ``source_fn`` result) is ``None`` are
+        silently skipped (the FB doesn't carry that field for the current
+        request).
+        """
+        ctx = FillContext(
+            raw_bs=raw_bs,
+            padded_bs=padded_bs,
+            raw_num_tokens=raw_num_tokens,
+            padded_num_tokens=padded_num_tokens,
+            pp_proxy_tensors=pp_proxy_tensors,
+        )
+
+        # Phase 1: reset padded regions where it matters.
+        for slot in self._slots.values():
+            if not slot.enabled or slot.buffer is None:
+                continue
+            raw_n = slot._raw_n(raw_bs, raw_num_tokens)
+            padded_n = slot._padded_n(padded_bs, padded_num_tokens)
+            slot.reset_padding(raw_n, padded_n)
+
+        # Phase 2: collect (dst, src) pairs and dispatch a grouped copy.
+        gpu_dsts: List[torch.Tensor] = []
+        gpu_srcs: List[torch.Tensor] = []
+        cpu_dsts: List[torch.Tensor] = []
+        cpu_srcs: List[torch.Tensor] = []
+        for slot in self._slots.values():
+            if not slot.enabled or slot.buffer is None or not slot.copy_from_fb:
+                continue
+            if slot.source_fn is not None:
+                # Structured / side-sourced slot: source comes from a nested FB
+                # dataclass or an out-of-band input, and the copy is sliced to
+                # the source's own length rather than a bs/tokens axis.
+                src = slot.source_fn(forward_batch, ctx)
+                if src is None:
+                    continue
+                dst = slot.buffer[: src.shape[0]]
+            else:
+                src = getattr(forward_batch, slot.name, None)
+                if src is None:
+                    continue
+                if not isinstance(src, torch.Tensor):
+                    # Non-tensor FB fields (e.g. dicts, dataclasses) are not
+                    # auto-copied — caller handles via source_fn or post_fill.
+                    continue
+                raw_n = slot._raw_n(raw_bs, raw_num_tokens)
+                if slot.slice_fn is not None:
+                    dst = slot.slice_fn(slot.buffer, raw_n)
+                elif slot.axis == "none":
+                    dst = slot.buffer
+                else:
+                    dst = slot.buffer[:raw_n]
+            # foreach_copy_ requires same-device tensors per call — bucket
+            # by device.
+            if dst.device.type == "cpu":
+                cpu_dsts.append(dst)
+                cpu_srcs.append(src)
+            else:
+                gpu_dsts.append(dst)
+                gpu_srcs.append(src)
+        if gpu_dsts:
+            _grouped_foreach_copy_(gpu_dsts, gpu_srcs)
+        for dst, src in zip(cpu_dsts, cpu_srcs):
+            dst.copy_(src)
+
+        # Phase 3: post-fill hooks (compute-then-write slots).
+        for slot in self._slots.values():
+            if not slot.enabled or slot.buffer is None or slot.post_fill is None:
+                continue
+            slot.post_fill(slot.buffer, forward_batch, ctx)
+
+    def extract_buffer(
+        self,
+        *,
+        padded_bs: int,
+        padded_num_tokens: int,
+        forward_batch_template: "ForwardBatch",
+    ) -> "ForwardBatch":
+        """Return a FB view backed by registry slot buffers.
+
+        ``forward_batch_template`` provides the non-slot fields
+        (``forward_mode`` / ``spec_info`` / ``sampling_info`` /
+        ``capture_hidden_mode`` / ``dp_*`` / ``lora_ids`` / ...). Slot
+        fields are replaced with views into the registry buffers via
+        ``dataclasses.replace`` — the template itself is not mutated.
+        """
+        import dataclasses
+
+        replace_kwargs: Dict[str, Any] = {"batch_size": padded_bs}
+        for slot in self._slots.values():
+            if not slot.enabled or slot.buffer is None:
+                continue
+            # Structured slots use dotted names ("<field>.<sub>") and are not
+            # top-level FB attributes — their data is consumed in place off the
+            # adopted backing object, not re-attached to the FB view here.
+            if "." in slot.name:
+                continue
+            replace_kwargs[slot.name] = slot.slice_for(padded_bs, padded_num_tokens)
+        return dataclasses.replace(forward_batch_template, **replace_kwargs)
+
+
+def build_decode_registry(
+    *,
+    device: torch.device,
+    max_bs: int,
+    max_num_token: int,
+    seq_len_fill_value: int,
+    cache_loc_dtype: torch.dtype,
+    enable_mamba_track: bool = False,
+    is_encoder_decoder: bool = False,
+    encoder_len_fill_value: int = 0,
+    enable_num_token_non_padded: bool = False,
+    require_gathered_buffer: bool = False,
+    enable_prefill_cp: bool = False,
+    require_mlp_tp_gather: bool = False,
+    dp_size: int = 1,
+    share_pool: bool = True,
+    source: Optional[Any] = None,
+) -> CudaGraphBufferRegistry:
+    """Registry mirroring the always-on (+ mamba / mrope) FB-shared decode
+    buffers, with padding policies matching
+    ``DecodeInputBuffers.populate_from_forward_batch``:
+
+      - ``seq_lens`` / ``seq_lens_cpu`` -> FILL_SENTINEL(seq_len_fill_value)
+      - ``req_pool_indices`` / ``out_cache_loc`` / ``mamba_track_*`` -> ZERO
+      - ``input_ids`` / ``positions`` / ``mrope_positions`` -> FOREACH_COPY
+        (head ``[:raw_n]`` is always overwritten by the copy; the old code's
+        full-buffer ``zero_()`` / ``fill_()`` on ``bs != raw_bs`` is therefore
+        equivalent to the tail-only reset the policies apply here).
+
+    ``custom_mask`` / ``next_token_logits_buffer`` / ``input_embeds`` are not
+    registered here — they are not per-replay FB copies (allocated and written
+    elsewhere), so the runner keeps owning them.
+
+    When ``source`` is given, each slot adopts the same-named tensor off
+    ``source`` (e.g. a ``DecodeInputBuffers``) instead of allocating, so the
+    registry shares one physical allocation with that object.
+    """
+    reg = CudaGraphBufferRegistry(
+        device=device,
+        max_bs=max_bs,
+        max_num_tokens=max_num_token,
+        share_pool=share_pool,
+    )
+
+    def _tokens(_bs: int, mt: int) -> Tuple[int, ...]:
+        return (mt,)
+
+    def _bs(bs: int, _mt: int) -> Tuple[int, ...]:
+        return (bs,)
+
+    slots = [
+        GraphSlot("input_ids", _tokens, torch.int64, axis="tokens"),
+        GraphSlot("positions", _tokens, torch.int64, axis="tokens"),
+        GraphSlot(
+            "out_cache_loc",
+            _tokens,
+            cache_loc_dtype,
+            axis="tokens",
+            padding_policy=PaddingPolicy.ZERO,
+        ),
+        GraphSlot(
+            "req_pool_indices",
+            _bs,
+            torch.int64,
+            axis="bs",
+            padding_policy=PaddingPolicy.ZERO,
+        ),
+        GraphSlot(
+            "seq_lens",
+            _bs,
+            torch.int32,
+            axis="bs",
+            padding_policy=PaddingPolicy.FILL_SENTINEL,
+            pad_value=seq_len_fill_value,
+        ),
+        GraphSlot(
+            "seq_lens_cpu",
+            _bs,
+            torch.int32,
+            axis="bs",
+            device=torch.device("cpu"),
+            padding_policy=PaddingPolicy.FILL_SENTINEL,
+            pad_value=seq_len_fill_value,
+        ),
+        GraphSlot(
+            "mrope_positions",
+            lambda _bs2, mt: (3, mt),
+            torch.int64,
+            axis="tokens",
+            slice_fn=lambda buf, n: buf[:, :n],
+        ),
+    ]
+    if enable_mamba_track:
+        slots.append(
+            GraphSlot(
+                "mamba_track_indices",
+                _bs,
+                torch.int64,
+                axis="bs",
+                padding_policy=PaddingPolicy.ZERO,
+            )
+        )
+        slots.append(
+            GraphSlot(
+                "mamba_track_mask",
+                _bs,
+                torch.bool,
+                axis="bs",
+                padding_policy=PaddingPolicy.ZERO,
+            )
+        )
+    if is_encoder_decoder:
+        # Initialized once to encoder_len_fill_value, copied head-only, never
+        # reset per iter — matching the legacy DecodeInputBuffers behavior.
+        slots.append(
+            GraphSlot(
+                "encoder_lens",
+                _bs,
+                torch.int32,
+                axis="bs",
+                padding_policy=PaddingPolicy.FILL_ONCE,
+                pad_value=encoder_len_fill_value,
+            )
+        )
+    if enable_num_token_non_padded:
+        from sglang.srt.model_executor.forward_batch_info import (
+            compute_local_num_token_non_padded,
+        )
+
+        def _num_token_non_padded_post_fill(buf, fb, ctx):
+            # Gathered (DP) path overwrites the plain FB copy with this rank's
+            # local count; the non-gathered path keeps the copied value.
+            if require_gathered_buffer and not enable_prefill_cp:
+                buf.copy_(
+                    compute_local_num_token_non_padded(
+                        global_num_token_non_padded=fb.num_token_non_padded,
+                        num_tokens_per_dp=ctx.padded_num_tokens,
+                    )
+                )
+
+        slots.append(
+            GraphSlot(
+                "num_token_non_padded",
+                lambda _bs, _mt: (1,),
+                torch.int32,
+                axis="none",
+                post_fill=_num_token_non_padded_post_fill,
+            )
+        )
+
+    def _global_num_tokens_post_fill(buf, fb, ctx):
+        # Filled with the padded token count on the gathered (DP) path; left
+        # untouched otherwise. Not an FB copy (copy_from_fb=False).
+        if require_gathered_buffer:
+            buf.fill_(ctx.padded_num_tokens)
+
+    _global_shape = (
+        (lambda _bs, _mt: (dp_size,))
+        if require_mlp_tp_gather
+        else (lambda _bs, _mt: (1,))
+    )
+    for _global_name in ("global_num_tokens_gpu", "global_num_tokens_for_logprob_gpu"):
+        slots.append(
+            GraphSlot(
+                _global_name,
+                _global_shape,
+                torch.int32,
+                axis="none",
+                copy_from_fb=False,
+                post_fill=_global_num_tokens_post_fill,
+            )
+        )
+
+    for slot in slots:
+        bind = None
+        if source is not None:
+            bind = getattr(source, slot.name, None)
+            if bind is None:
+                raise ValueError(
+                    f"source is missing buffer {slot.name!r} required by the "
+                    "decode registry; cannot adopt."
+                )
+        reg.register_slot(slot, bind=bind)
+
+    # Structured slots whose backing storage still lives on the source object
+    # (adopt-only during migration): registered only when the source actually
+    # carries them. The per-replay copy source is a nested FB dataclass field,
+    # supplied via source_fn; head is copied (source-length slice), tail kept.
+    if source is not None:
+        ngram = getattr(source, "ngram_embedding_info", None)
+        if ngram is not None:
+
+            def _ngram_source(attr):
+                def _fn(fb, _ctx):
+                    info = getattr(fb, "ngram_embedding_info", None)
+                    return None if info is None else getattr(info, attr)
+
+                return _fn
+
+            for _attr in ("column_starts", "req_lens"):
+                backing = getattr(ngram, _attr)
+                reg.register_slot(
+                    GraphSlot(
+                        name=f"ngram_embedding_info.{_attr}",
+                        shape_fn=lambda _bs, _mt, _s=tuple(backing.shape): _s,
+                        dtype=backing.dtype,
+                        axis="none",
+                        padding_policy=PaddingPolicy.KEEP_PAD,
+                        source_fn=_ngram_source(_attr),
+                    ),
+                    bind=backing,
+                )
+
+        # Pipeline-parallel proxy tensors: a dict of per-key buffers, sourced
+        # from the out-of-band pp input on FillContext rather than the FB.
+        pp = getattr(source, "pp_proxy_tensors", None)
+        if pp is not None:
+
+            def _pp_source(key):
+                def _fn(_fb, ctx):
+                    ppx = ctx.pp_proxy_tensors
+                    return None if ppx is None else ppx.tensors[key]
+
+                return _fn
+
+            for _key, _backing in pp.items():
+                reg.register_slot(
+                    GraphSlot(
+                        name=f"pp_proxy_tensors.{_key}",
+                        shape_fn=lambda _bs, _mt, _s=tuple(_backing.shape): _s,
+                        dtype=_backing.dtype,
+                        axis="none",
+                        padding_policy=PaddingPolicy.KEEP_PAD,
+                        source_fn=_pp_source(_key),
+                    ),
+                    bind=_backing,
+                )
+
+        # KV-canary id buffers (off by default): plain bs-axis FB copies,
+        # adopt-only when the source carries them. Head [:raw_bs] is copied;
+        # the tail keeps its init (rids_int 0, bootstrap_room_ids_int -1).
+        for _cname in ("rids_int", "bootstrap_room_ids_int"):
+            canary = getattr(source, _cname, None)
+            if canary is not None:
+                reg.register_slot(
+                    GraphSlot(
+                        name=_cname,
+                        shape_fn=lambda _bs, _mt, _s=tuple(canary.shape): _s,
+                        dtype=canary.dtype,
+                        axis="bs",
+                    ),
+                    bind=canary,
+                )
+
+    return reg
+
+
+def build_prefill_registry(
+    *,
+    device: torch.device,
+    max_bs: int,
+    max_num_token: int,
+    cache_loc_dtype: torch.dtype,
+    is_multimodal: bool = False,
+    hidden_size: int = 0,
+    embed_dtype: Optional[torch.dtype] = None,
+    enable_mamba_track: bool = False,
+    share_pool: bool = True,
+    source: Optional[Any] = None,
+) -> CudaGraphBufferRegistry:
+    """Registry mirroring the **token-axis** FB-shared buffers for the
+    piecewise / breakable (prefill) cuda-graph runners.
+
+    Padding policies match the inline copy/zero in
+    ``PiecewiseCudaGraphRunner.replay_prepare``: ``input_ids`` / ``positions``
+    / ``out_cache_loc`` / ``mrope_positions`` / ``input_embeds`` reset their
+    padded tail ``[raw_num_tokens:padded_num_tokens]`` to ``0`` (the padded
+    tokens *are* processed by the graph, so they must be benign), then the head
+    ``[:raw_num_tokens]`` is copied from the FB. ``input_embeds`` is not an FB
+    copy — the model writes the embeds into it inside the graph — so it is
+    reset-only (``copy_from_fb=False``). ``mamba_track_*`` are bs-axis copies
+    with no padding reset (bs is not padded on this path).
+
+    When ``source`` is given, each slot adopts the same-named tensor off
+    ``source`` (the ``PrefillInputBuffers``) instead of allocating, so the
+    registry shares one physical allocation (and ``data_ptr``) with it.
+    """
+    reg = CudaGraphBufferRegistry(
+        device=device,
+        max_bs=max_bs,
+        max_num_tokens=max_num_token,
+        share_pool=share_pool,
+    )
+
+    def _tokens(_bs: int, mt: int) -> Tuple[int, ...]:
+        return (mt,)
+
+    def _bs(bs: int, _mt: int) -> Tuple[int, ...]:
+        return (bs,)
+
+    slots = [
+        GraphSlot(
+            "input_ids",
+            _tokens,
+            torch.int64,
+            axis="tokens",
+            padding_policy=PaddingPolicy.ZERO,
+        ),
+        GraphSlot(
+            "positions",
+            _tokens,
+            torch.int64,
+            axis="tokens",
+            padding_policy=PaddingPolicy.ZERO,
+        ),
+        GraphSlot(
+            "out_cache_loc",
+            _tokens,
+            cache_loc_dtype,
+            axis="tokens",
+            padding_policy=PaddingPolicy.ZERO,
+        ),
+    ]
+    if is_multimodal:
+        slots.append(
+            GraphSlot(
+                "mrope_positions",
+                lambda _bs2, mt: (3, mt),
+                torch.int64,
+                axis="tokens",
+                padding_policy=PaddingPolicy.ZERO,
+                slice_fn=lambda buf, n: buf[:, :n],
+            )
+        )
+        slots.append(
+            GraphSlot(
+                "input_embeds",
+                lambda _bs2, mt: (mt, hidden_size),
+                embed_dtype,
+                axis="tokens",
+                padding_policy=PaddingPolicy.ZERO,
+                copy_from_fb=False,
+            )
+        )
+    if enable_mamba_track:
+        slots.append(GraphSlot("mamba_track_indices", _bs, torch.int64, axis="bs"))
+        slots.append(GraphSlot("mamba_track_mask", _bs, torch.bool, axis="bs"))
+        slots.append(GraphSlot("mamba_track_seqlens", _bs, torch.int32, axis="bs"))
+
+    for slot in slots:
+        bind = None
+        if source is not None:
+            bind = getattr(source, slot.name, None)
+            if bind is None:
+                raise ValueError(
+                    f"source is missing buffer {slot.name!r} required by the "
+                    "prefill registry; cannot adopt."
+                )
+        reg.register_slot(slot, bind=bind)
+    return reg
diff --git a/python/sglang/srt/model_executor/cuda_graph_runner.py b/python/sglang/srt/model_executor/cuda_graph_runner.py
index aa88d7c99..300cd9a84 100644
--- a/python/sglang/srt/model_executor/cuda_graph_runner.py
+++ b/python/sglang/srt/model_executor/cuda_graph_runner.py
@@ -25,7 +25,7 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import partial
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Callable, Dict, Optional, Union
 
 import torch
 import tqdm
@@ -58,6 +58,7 @@ from sglang.srt.layers.moe.token_dispatcher.deepep import DeepEPBuffer
 from sglang.srt.layers.moe.utils import get_deepep_mode, get_moe_a2a_backend
 from sglang.srt.layers.utils import MultiPlatformOp
 from sglang.srt.layers.utils.cp_utils import is_mla_prefill_cp_enabled
+from sglang.srt.model_executor.cuda_graph_buffer_registry import build_decode_registry
 from sglang.srt.model_executor.forward_batch_info import (
     CaptureHiddenMode,
     ForwardBatch,
@@ -105,8 +106,6 @@ logger = logging.getLogger(__name__)
 if TYPE_CHECKING:
     from sglang.srt.model_executor.model_runner import ModelRunner
 
-_has_foreach_copy = hasattr(torch, "_foreach_copy_")
-
 
 def build_replay_fb_view(
     forward_batch: "ForwardBatch",
@@ -159,27 +158,6 @@ def build_replay_fb_view(
     )
 
 
-def _grouped_foreach_copy_(dsts: List[torch.Tensor], srcs: List[torch.Tensor]) -> None:
-    """Call torch._foreach_copy_ grouped by (dst_dtype, src_dtype) pairs."""
-
-    def foreach_copy(dsts: List[torch.Tensor], srcs: List[torch.Tensor]) -> None:
-        if _has_foreach_copy:
-            torch._foreach_copy_(dsts, srcs)
-        else:
-            for dst, src in zip(dsts, srcs):
-                dst.copy_(src)
-
-    groups: Dict[Tuple[torch.dtype, torch.dtype], Tuple[List, List]] = {}
-    for dst, src in zip(dsts, srcs):
-        key = (dst.dtype, src.dtype)
-        if key not in groups:
-            groups[key] = ([], [])
-        groups[key][0].append(dst)
-        groups[key][1].append(src)
-    for group_dsts, group_srcs in groups.values():
-        foreach_copy(group_dsts, group_srcs)
-
-
 @dataclass
 class DecodeInputBuffers(ForwardInputBuffers):
 
@@ -344,110 +322,19 @@ class DecodeInputBuffers(ForwardInputBuffers):
         num_tokens_per_bs: int,
         dsa_enable_prefill_cp: bool,
         enable_num_token_non_padded_flag: bool,
+        registry,
         pp_proxy_tensors: Optional[PPProxyTensors] = None,
     ):
-        if bs != raw_bs:
-            self.seq_lens.fill_(seq_len_fill_value)
-            self.out_cache_loc.zero_()
-            # Pair with seq_lens fill: padded rows must point at reserved
-            # req_pool slot 0 (req_to_token[0, :] is all zeros from init),
-            # so dummy attention reads land on slot 0 instead of a stale
-            # req_to_token row left by an earlier replay.
-            self.req_pool_indices.zero_()
-            if self.mamba_track_indices is not None:
-                self.mamba_track_indices.zero_()
-            if self.mamba_track_mask is not None:
-                self.mamba_track_mask.fill_(False)
-
-        # Build batched copy lists for all GPU tensors.
-        dsts = [
-            self.input_ids[:raw_num_token],
-            self.req_pool_indices[:raw_bs],
-            self.seq_lens[:raw_bs],
-            self.out_cache_loc[:raw_num_token],
-            self.positions[:raw_num_token],
-        ]
-        srcs = [
-            forward_batch.input_ids,
-            forward_batch.req_pool_indices,
-            forward_batch.seq_lens,
-            forward_batch.out_cache_loc,
-            forward_batch.positions,
-        ]
-
-        if self.ngram_embedding_info is not None:
-            ngram_embedding_info = forward_batch.ngram_embedding_info
-            self.ngram_embedding_info.column_starts[:raw_bs].copy_(
-                ngram_embedding_info.column_starts
-            )
-            self.ngram_embedding_info.req_lens[:raw_bs].copy_(
-                ngram_embedding_info.req_lens
-            )
-
-        if (
-            self.mamba_track_indices is not None
-            and forward_batch.mamba_track_indices is not None
-        ):
-            dsts.append(self.mamba_track_indices[:raw_bs])
-            srcs.append(forward_batch.mamba_track_indices)
-        if (
-            self.mamba_track_mask is not None
-            and forward_batch.mamba_track_mask is not None
-        ):
-            dsts.append(self.mamba_track_mask[:raw_bs])
-            srcs.append(forward_batch.mamba_track_mask)
-
-        if self.encoder_lens is not None and forward_batch.encoder_lens is not None:
-            dsts.append(self.encoder_lens[:raw_bs])
-            srcs.append(forward_batch.encoder_lens)
-
-        if forward_batch.mrope_positions is not None:
-            dsts.append(self.mrope_positions[:, :raw_num_token])
-            srcs.append(forward_batch.mrope_positions)
-
-        if self.rids_int is not None and forward_batch.rids_int is not None:
-            dsts.append(self.rids_int[:raw_bs])
-            srcs.append(forward_batch.rids_int)
-        if (
-            self.bootstrap_room_ids_int is not None
-            and forward_batch.bootstrap_room_ids_int is not None
-        ):
-            dsts.append(self.bootstrap_room_ids_int[:raw_bs])
-            srcs.append(forward_batch.bootstrap_room_ids_int)
-
-        if require_gathered_buffer:
-            self.global_num_tokens_gpu.fill_(bs * num_tokens_per_bs)
-            self.global_num_tokens_for_logprob_gpu.fill_(bs * num_tokens_per_bs)
-
-        if enable_num_token_non_padded_flag:
-            if require_gathered_buffer and not dsa_enable_prefill_cp:
-                num_tokens_per_dp = bs * num_tokens_per_bs
-                local = compute_local_num_token_non_padded(
-                    global_num_token_non_padded=forward_batch.num_token_non_padded,
-                    num_tokens_per_dp=num_tokens_per_dp,
-                )
-                dsts.append(self.num_token_non_padded)
-                srcs.append(local)
-            else:
-                dsts.append(self.num_token_non_padded)
-                srcs.append(forward_batch.num_token_non_padded)
-
-        # Pipeline-parallel proxy tensors.
-        if pp_proxy_tensors is not None and self.pp_proxy_tensors is not None:
-            for key, buf in self.pp_proxy_tensors.items():
-                src = pp_proxy_tensors.tensors[key]
-                dim = src.shape[0]
-                dsts.append(buf[:dim])
-                srcs.append(src)
-
-        # Batch all GPU copies, grouped by dtype pair.
-        _grouped_foreach_copy_(dsts, srcs)
-
-        # CPU tensor copy (cannot be batched with GPU tensors).
-        if forward_batch.seq_lens_cpu is not None:
-            if bs != raw_bs:
-                self.seq_lens_cpu.fill_(seq_len_fill_value)
-            self.seq_lens_cpu[:raw_bs].copy_(forward_batch.seq_lens_cpu)
+        # Reset padded tails + copy FB into the registry-adopted graph buffers
+        # (same storage the old per-field populate wrote).
+        registry.fill_from(
+            forward_batch,
+            raw_bs=raw_bs,
+            padded_bs=bs,
+            raw_num_tokens=raw_num_token,
+            padded_num_tokens=bs * num_tokens_per_bs,
+            pp_proxy_tensors=pp_proxy_tensors,
+        )
 
 
 # Detect whether the current forward pass is in capture mode
@@ -767,6 +654,26 @@ class CudaGraphRunner:
             ),
         )
         self.buffers.share_buffers()
+        # FB-shared slot registry, adopting the DecodeInputBuffers storage so
+        # it mirrors the same physical buffers (stable data_ptr for capture vs
+        # replay). This is the unified fill/extract surface that eager /
+        # capture / replay migrate onto, replacing populate_from_forward_batch.
+        self.buffer_registry = build_decode_registry(
+            device=self.device,
+            max_bs=self.max_bs,
+            max_num_token=self.max_num_token,
+            seq_len_fill_value=self.seq_len_fill_value,
+            cache_loc_dtype=self._cache_loc_dtype(),
+            enable_mamba_track=enable_mamba_track,
+            is_encoder_decoder=self.is_encoder_decoder,
+            encoder_len_fill_value=self.encoder_len_fill_value,
+            enable_num_token_non_padded=enable_num_token_non_padded(),
+            require_gathered_buffer=self.require_gathered_buffer,
+            enable_prefill_cp=self.enable_prefill_cp,
+            require_mlp_tp_gather=self.require_mlp_tp_gather,
+            dp_size=self.dp_size,
+            source=self.buffers,
+        )
 
         self.tbo_plugin = TboCudaGraphRunnerPlugin()
 
@@ -995,18 +902,24 @@ class CudaGraphRunner:
         stream = self.stream
         num_tokens = bs * self.num_tokens_per_bs
 
-        # Graph inputs
-        input_ids = buffers.input_ids[:num_tokens]
-        req_pool_indices = buffers.req_pool_indices[:bs]
-        seq_lens = buffers.seq_lens[:bs]
-        seq_lens_cpu = buffers.seq_lens_cpu[:bs]
-        out_cache_loc = buffers.out_cache_loc[:num_tokens]
-        positions = buffers.positions[:num_tokens]
-        if self.is_encoder_decoder:
-            encoder_lens = buffers.encoder_lens[:bs]
-        else:
-            encoder_lens = None
-        mrope_positions = buffers.mrope_positions[:, :num_tokens]
+        # Graph inputs. The registry-owned FB-shared slots come from the
+        # registry (it adopted the DecodeInputBuffers storage, so these are the
+        # same physical tensors); the rest still come off `buffers` directly.
+        registry = self.buffer_registry
+
+        def _slot(name):
+            return registry.get_slot(name).slice_for(bs, num_tokens)
+
+        input_ids = _slot("input_ids")
+        req_pool_indices = _slot("req_pool_indices")
+        seq_lens = _slot("seq_lens")
+        seq_lens_cpu = _slot("seq_lens_cpu")
+        out_cache_loc = _slot("out_cache_loc")
+        positions = _slot("positions")
+        encoder_lens = (
+            _slot("encoder_lens") if registry.has_slot("encoder_lens") else None
+        )
+        mrope_positions = _slot("mrope_positions")
         next_token_logits_buffer = buffers.next_token_logits_buffer[:num_tokens]
         rids_int = buffers.rids_int[:bs] if buffers.rids_int is not None else None
         bootstrap_room_ids_int = (
@@ -1065,16 +978,14 @@ class CudaGraphRunner:
         else:
             lora_ids = None
 
-        # mamba state tracking
+        # mamba state tracking (registry-owned when enabled)
         mamba_track_indices = (
-            buffers.mamba_track_indices[:bs]
-            if buffers.mamba_track_indices is not None
+            _slot("mamba_track_indices")
+            if registry.has_slot("mamba_track_indices")
             else None
         )
         mamba_track_mask = (
-            buffers.mamba_track_mask[:bs]
-            if buffers.mamba_track_mask is not None
-            else None
+            _slot("mamba_track_mask") if registry.has_slot("mamba_track_mask") else None
         )
 
         if stream_idx is None:
@@ -1275,6 +1186,7 @@ class CudaGraphRunner:
             # "any prefill-CP flavor enabled" (DSA CP or MLA CP).
             dsa_enable_prefill_cp=self.enable_prefill_cp,
             enable_num_token_non_padded_flag=enable_num_token_non_padded(),
+            registry=self.buffer_registry,
             pp_proxy_tensors=pp_proxy_tensors,
         )
 
diff --git a/python/sglang/srt/model_executor/input_buffers.py b/python/sglang/srt/model_executor/input_buffers.py
index b9d123e58..efb1f4255 100644
--- a/python/sglang/srt/model_executor/input_buffers.py
+++ b/python/sglang/srt/model_executor/input_buffers.py
@@ -2,36 +2,42 @@ from __future__ import annotations
 
 import dataclasses
 from dataclasses import dataclass, fields
-from typing import Dict
+from typing import Dict, Tuple
 
 import torch
 
 from sglang.srt.utils import is_npu
 
-_forward_input_buffer_pool: Dict[str, torch.Tensor] = {}
+# Process-wide pool keyed by (name, numel, dtype, device); see share_input_buffer.
+_PoolKey = Tuple[str, int, torch.dtype, torch.device]
+_forward_input_buffer_pool: Dict[_PoolKey, torch.Tensor] = {}
+
+
+def share_input_buffer(name: str, new_buffer: torch.Tensor) -> torch.Tensor:
+    """Coalesce a buffer by ``(name, size, dtype, device)`` into the
+    process-wide input-buffer pool.
+
+    Distinct callers that request the same field ``name`` with the same
+    size/dtype/device share one physical allocation (and therefore one
+    ``data_ptr``): the first registrant's buffer becomes canonical and every
+    later identical request is returned as a view aliased onto it. Requests
+    that differ in size get their own allocation — they never reuse or displace
+    an existing entry — so the sharing *structure* is independent of
+    registration order and no already-captured buffer is ever repointed.
+    """
+    key: _PoolKey = (name, new_buffer.numel(), new_buffer.dtype, new_buffer.device)
+    canonical = _forward_input_buffer_pool.get(key, None)
+    if canonical is None:
+        _forward_input_buffer_pool[key] = new_buffer
+        canonical = new_buffer
+    return canonical.as_strided(new_buffer.size(), new_buffer.stride())
 
 
 @dataclass
 class ForwardInputBuffers:
 
     def _share_one_buffer(self, name: str, new_buffer: torch.Tensor) -> torch.Tensor:
-
-        buffer_size = new_buffer.size()
-        buffer_stride = new_buffer.stride()
-
-        old_buffer = _forward_input_buffer_pool.get(name, None)
-        if old_buffer is not None:
-            assert (
-                new_buffer.dtype == old_buffer.dtype
-            ), f"Buffer {name} has different dtype than before."
-            assert (
-                new_buffer.device == old_buffer.device
-            ), f"Buffer {name} has different device than before."
-            if old_buffer.numel() > new_buffer.numel():
-                new_buffer = old_buffer
-
-        _forward_input_buffer_pool[name] = new_buffer
-        return new_buffer.as_strided(buffer_size, buffer_stride)
+        return share_input_buffer(name, new_buffer)
 
     def share_buffers(self):
         # disable share input buffer on npu due to accuracy issue
diff --git a/python/sglang/srt/model_executor/piecewise_cuda_graph_runner.py b/python/sglang/srt/model_executor/piecewise_cuda_graph_runner.py
index 0ba8d5d99..b6ed9ac07 100644
--- a/python/sglang/srt/model_executor/piecewise_cuda_graph_runner.py
+++ b/python/sglang/srt/model_executor/piecewise_cuda_graph_runner.py
@@ -52,6 +52,7 @@ from sglang.srt.layers.logits_processor import LogitsProcessorOutput
 from sglang.srt.layers.moe.utils import get_moe_a2a_backend
 from sglang.srt.layers.pooler import EmbeddingPoolerOutput
 from sglang.srt.layers.utils import MultiPlatformOp
+from sglang.srt.model_executor.cuda_graph_buffer_registry import build_prefill_registry
 from sglang.srt.model_executor.forward_batch_info import (
     CaptureHiddenMode,
     ForwardBatch,
@@ -295,6 +296,20 @@ class PiecewiseCudaGraphRunner:
         )
         self.buffers.share_buffers()
 
+        # Token-axis FB-shared slot registry, adopting the PrefillInputBuffers
+        # storage (one data_ptr shared with capture + replay).
+        self.buffer_registry = build_prefill_registry(
+            device=self.device,
+            max_bs=self.max_bs,
+            max_num_token=self.max_num_tokens,
+            cache_loc_dtype=self._cache_loc_dtype(),
+            is_multimodal=self.is_multimodal,
+            hidden_size=self.model_runner.model_config.hidden_size,
+            embed_dtype=self.model_runner.dtype,
+            enable_mamba_track=self.mamba_track_enabled,
+            source=self.buffers,
+        )
+
         self.attention_layers = self.model_runner.attention_layers
         self.moe_layers = self.model_runner.moe_layers
         self.moe_fusions = self.model_runner.moe_fusions
@@ -355,27 +370,32 @@ class PiecewiseCudaGraphRunner:
 
     def warmup_compile(self, num_tokens: int):
         """Warmup the model with a simple forward pass before CUDA graph capture."""
-        buffers = self.buffers
-        input_ids = buffers.input_ids[:num_tokens]
-        input_embeds = buffers.input_embeds[:num_tokens] if self.is_multimodal else None
-        positions = buffers.positions[:num_tokens]
+        registry = self.buffer_registry
+        bs = 1
+
+        def _slot(name):
+            return registry.get_slot(name).slice_for(bs, num_tokens)
+
+        input_ids = _slot("input_ids")
+        positions = _slot("positions")
+        out_cache_loc = _slot("out_cache_loc")
+        input_embeds = (
+            _slot("input_embeds") if registry.has_slot("input_embeds") else None
+        )
         mrope_positions = (
-            buffers.mrope_positions[:, :num_tokens] if self.is_multimodal else None
+            _slot("mrope_positions") if registry.has_slot("mrope_positions") else None
         )
-        out_cache_loc = buffers.out_cache_loc[:num_tokens]
         mamba_track_indices = (
-            buffers.mamba_track_indices[:1]
-            if buffers.mamba_track_indices is not None
+            _slot("mamba_track_indices")
+            if registry.has_slot("mamba_track_indices")
             else None
         )
         mamba_track_mask = (
-            buffers.mamba_track_mask[:1]
-            if buffers.mamba_track_mask is not None
-            else None
+            _slot("mamba_track_mask") if registry.has_slot("mamba_track_mask") else None
         )
         mamba_track_seqlens = (
-            buffers.mamba_track_seqlens[:1]
-            if buffers.mamba_track_seqlens is not None
+            _slot("mamba_track_seqlens")
+            if registry.has_slot("mamba_track_seqlens")
             else None
         )
         with torch.device(self.device):
@@ -508,33 +528,36 @@ class PiecewiseCudaGraphRunner:
                     self.capture_one_batch_size(num_tokens)
 
     def capture_one_batch_size(self, num_tokens: int):
-        buffers = self.buffers
+        registry = self.buffer_registry
         bs = 1
 
-        # Graph inputs
-        input_ids = buffers.input_ids[:num_tokens]
-        input_embeds = buffers.input_embeds[:num_tokens] if self.is_multimodal else None
+        # Graph inputs — views into the registry's (adopted) graph-resident
+        # slots; capture burns these addresses into the graph.
+        def _slot(name):
+            return registry.get_slot(name).slice_for(bs, num_tokens)
 
-        out_cache_loc = buffers.out_cache_loc[:num_tokens]
+        input_ids = _slot("input_ids")
+        positions = _slot("positions")
+        out_cache_loc = _slot("out_cache_loc")
+        input_embeds = (
+            _slot("input_embeds") if registry.has_slot("input_embeds") else None
+        )
+        mrope_positions = (
+            _slot("mrope_positions") if registry.has_slot("mrope_positions") else None
+        )
         mamba_track_indices = (
-            buffers.mamba_track_indices[:bs]
-            if buffers.mamba_track_indices is not None
+            _slot("mamba_track_indices")
+            if registry.has_slot("mamba_track_indices")
             else None
         )
         mamba_track_mask = (
-            buffers.mamba_track_mask[:bs]
-            if buffers.mamba_track_mask is not None
-            else None
+            _slot("mamba_track_mask") if registry.has_slot("mamba_track_mask") else None
         )
         mamba_track_seqlens = (
-            buffers.mamba_track_seqlens[:bs]
-            if buffers.mamba_track_seqlens is not None
+            _slot("mamba_track_seqlens")
+            if registry.has_slot("mamba_track_seqlens")
             else None
         )
-        positions = buffers.positions[:num_tokens]
-        mrope_positions = (
-            buffers.mrope_positions[:, :num_tokens] if self.is_multimodal else None
-        )
 
         global_dp_buffer_len = None
         global_num_tokens_cpu = None
@@ -649,72 +672,51 @@ class PiecewiseCudaGraphRunner:
         forward_batch: ForwardBatch,
         **kwargs,
     ):
-        buffers = self.buffers
         num_tokens = len(forward_batch.input_ids)
         index = bisect.bisect_left(self.capture_num_tokens, num_tokens)
         static_num_tokens = self.capture_num_tokens[index]
         self.raw_num_tokens = num_tokens
-        if static_num_tokens != num_tokens:
-            buffers.out_cache_loc.zero_()
-            buffers.input_ids[num_tokens:static_num_tokens].zero_()
-            buffers.positions[num_tokens:static_num_tokens].zero_()
-            if self.is_multimodal:
-                buffers.input_embeds[num_tokens:static_num_tokens].zero_()
-            if forward_batch.mrope_positions is not None:
-                buffers.mrope_positions[:, num_tokens:static_num_tokens].zero_()
-
         bs = forward_batch.batch_size
+        registry = self.buffer_registry
+        # Reset the padded token tail (ZERO) + copy the [:num_tokens] head for
+        # every graph-resident slot in one grouped pass. input_embeds is
+        # reset-only (the model writes embeds into it inside the graph).
+        registry.fill_from(
+            forward_batch,
+            raw_bs=bs,
+            padded_bs=bs,
+            raw_num_tokens=num_tokens,
+            padded_num_tokens=static_num_tokens,
+        )
 
-        buffers.input_ids[:num_tokens].copy_(forward_batch.input_ids)
-        buffers.positions[:num_tokens].copy_(forward_batch.positions)
-        buffers.out_cache_loc[:num_tokens].copy_(forward_batch.out_cache_loc)
-
-        if (
-            buffers.mamba_track_indices is not None
-            and forward_batch.mamba_track_indices is not None
-        ):
-            buffers.mamba_track_indices[:bs].copy_(forward_batch.mamba_track_indices)
-        if (
-            buffers.mamba_track_mask is not None
-            and forward_batch.mamba_track_mask is not None
-        ):
-            buffers.mamba_track_mask[:bs].copy_(forward_batch.mamba_track_mask)
-        if (
-            buffers.mamba_track_seqlens is not None
-            and forward_batch.mamba_track_seqlens is not None
-        ):
-            buffers.mamba_track_seqlens[:bs].copy_(forward_batch.mamba_track_seqlens)
-
-        input_ids = buffers.input_ids[:static_num_tokens]
-        positions = buffers.positions[:static_num_tokens]
-        out_cache_loc = buffers.out_cache_loc[:static_num_tokens]
+        def _slot(name):
+            return registry.get_slot(name).slice_for(bs, static_num_tokens)
 
+        input_ids = _slot("input_ids")
+        positions = _slot("positions")
+        out_cache_loc = _slot("out_cache_loc")
         mamba_track_indices = (
-            buffers.mamba_track_indices[:bs]
-            if buffers.mamba_track_indices is not None
+            _slot("mamba_track_indices")
+            if registry.has_slot("mamba_track_indices")
             else None
         )
         mamba_track_mask = (
-            buffers.mamba_track_mask[:bs]
-            if buffers.mamba_track_mask is not None
-            else None
+            _slot("mamba_track_mask") if registry.has_slot("mamba_track_mask") else None
         )
         mamba_track_seqlens = (
-            buffers.mamba_track_seqlens[:bs]
-            if buffers.mamba_track_seqlens is not None
+            _slot("mamba_track_seqlens")
+            if registry.has_slot("mamba_track_seqlens")
             else None
         )
-        if forward_batch.mrope_positions is not None:
-            buffers.mrope_positions[:, :num_tokens].copy_(forward_batch.mrope_positions)
-
-        input_ids = buffers.input_ids[:static_num_tokens]
         input_embeds = (
-            buffers.input_embeds[:static_num_tokens] if self.is_multimodal else None
+            _slot("input_embeds") if registry.has_slot("input_embeds") else None
         )
-
         mrope_positions = (
-            buffers.mrope_positions[:, :static_num_tokens]
-            if forward_batch.mrope_positions is not None
+            _slot("mrope_positions")
+            if (
+                registry.has_slot("mrope_positions")
+                and forward_batch.mrope_positions is not None
+            )
             else None
         )
 
diff --git a/test/registered/unit/model_executor/test_cuda_graph_buffer_registry.py b/test/registered/unit/model_executor/test_cuda_graph_buffer_registry.py
new file mode 100644
index 000000000..602d3f9c3
--- /dev/null
+++ b/test/registered/unit/model_executor/test_cuda_graph_buffer_registry.py
@@ -0,0 +1,1201 @@
+"""Unit tests for ``cuda_graph_buffer_registry`` — CPU-only.
+
+Covers:
+  * ``GraphSlot`` shape / axis / device validation.
+  * ``register_slot`` allocation + ``FILL_SENTINEL`` init.
+  * ``fill_from`` D2D copy across all four ``PaddingPolicy`` modes.
+  * ``extract_buffer`` returns a ``ForwardBatch`` view backed by slot
+    buffers, non-slot fields carried from template.
+  * ``post_fill`` hook runs after the grouped copy.
+  * Missing FB attributes are silently skipped.
+
+The registry is GPU-agnostic — tests run on CPU. The PaddingPolicy /
+foreach_copy / view-slicing logic is fully exercised without any CUDA
+context.
+"""
+
+import dataclasses
+import unittest
+from types import SimpleNamespace
+from typing import Optional
+
+import torch
+
+from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+    CudaGraphBufferRegistry,
+    GraphSlot,
+    PaddingPolicy,
+)
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=10, suite="base-a-test-cpu")
+
+
+@dataclasses.dataclass
+class _MiniForwardBatch:
+    """Minimal FB stand-in: dataclass so ``dataclasses.replace`` works."""
+
+    batch_size: int = 0
+    input_ids: Optional[torch.Tensor] = None
+    seq_lens: Optional[torch.Tensor] = None
+    req_pool_indices: Optional[torch.Tensor] = None
+    out_cache_loc: Optional[torch.Tensor] = None
+    positions: Optional[torch.Tensor] = None
+    seq_lens_cpu: Optional[torch.Tensor] = None
+    encoder_lens: Optional[torch.Tensor] = None
+    mrope_positions: Optional[torch.Tensor] = None
+    num_token_non_padded: Optional[torch.Tensor] = None
+    global_num_tokens_gpu: Optional[torch.Tensor] = None
+    global_num_tokens_for_logprob_gpu: Optional[torch.Tensor] = None
+    ngram_embedding_info: Optional[object] = None
+    rids_int: Optional[torch.Tensor] = None
+    bootstrap_room_ids_int: Optional[torch.Tensor] = None
+    input_embeds: Optional[torch.Tensor] = None
+    mamba_track_indices: Optional[torch.Tensor] = None
+    mamba_track_mask: Optional[torch.Tensor] = None
+    mamba_track_seqlens: Optional[torch.Tensor] = None
+    forward_mode: Optional[str] = None
+    spec_info: Optional[object] = None
+
+
+def _make_registry(max_bs: int = 8, max_num_tokens: int = 16):
+    return CudaGraphBufferRegistry(
+        device=torch.device("cpu"),
+        max_bs=max_bs,
+        max_num_tokens=max_num_tokens,
+    )
+
+
+class TestGraphSlot(unittest.TestCase):
+    def test_axis_validation(self):
+        with self.assertRaises(ValueError):
+            GraphSlot(
+                name="bad",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="garbage",
+            )
+
+    def test_slice_for_before_buffer_alloc_raises(self):
+        slot = GraphSlot(
+            name="x",
+            shape_fn=lambda bs, mt: (bs,),
+            dtype=torch.int32,
+            axis="bs",
+        )
+        with self.assertRaises(RuntimeError):
+            slot.slice_for(padded_bs=1, padded_num_tokens=1)
+
+
+class TestRegistryRegister(unittest.TestCase):
+    def test_register_allocates_zero_buffer(self):
+        r = _make_registry()
+        slot = r.register_slot(
+            GraphSlot(
+                name="input_ids",
+                shape_fn=lambda bs, mt: (mt,),
+                dtype=torch.int64,
+                axis="tokens",
+            )
+        )
+        self.assertEqual(slot.buffer.shape, (16,))
+        self.assertEqual(slot.buffer.dtype, torch.int64)
+        self.assertTrue(torch.equal(slot.buffer, torch.zeros(16, dtype=torch.int64)))
+
+    def test_register_fill_sentinel_init(self):
+        r = _make_registry()
+        slot = r.register_slot(
+            GraphSlot(
+                name="seq_lens",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+                padding_policy=PaddingPolicy.FILL_SENTINEL,
+                pad_value=7,
+            )
+        )
+        self.assertTrue(
+            torch.equal(slot.buffer, torch.full((8,), 7, dtype=torch.int32))
+        )
+
+    def test_register_duplicate_raises(self):
+        r = _make_registry()
+        r.register_slot(
+            GraphSlot(
+                name="dup",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+            )
+        )
+        with self.assertRaises(ValueError):
+            r.register_slot(
+                GraphSlot(
+                    name="dup",
+                    shape_fn=lambda bs, mt: (bs,),
+                    dtype=torch.int32,
+                    axis="bs",
+                )
+            )
+
+    def test_disabled_slot_not_allocated(self):
+        r = _make_registry()
+        slot = r.register_slot(
+            GraphSlot(
+                name="off",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+                enabled=False,
+            )
+        )
+        self.assertIsNone(slot.buffer)
+        self.assertFalse(r.has_slot("off"))
+        self.assertNotIn("off", r.slot_names())
+
+    def test_cpu_device_override(self):
+        r = _make_registry()
+        slot = r.register_slot(
+            GraphSlot(
+                name="seq_lens_cpu",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+                device=torch.device("cpu"),
+                padding_policy=PaddingPolicy.FILL_SENTINEL,
+                pad_value=11,
+            )
+        )
+        self.assertEqual(slot.buffer.device.type, "cpu")
+        self.assertEqual(int(slot.buffer[0].item()), 11)
+
+
+class TestFillFromAndExtract(unittest.TestCase):
+    """End-to-end exercise: register a representative slot set, fill from
+    a mini FB, then extract a FB view and assert all field-views match."""
+
+    def _build_registry(self):
+        r = _make_registry(max_bs=4, max_num_tokens=8)
+        r.register_slot(
+            GraphSlot(
+                name="input_ids",
+                shape_fn=lambda bs, mt: (mt,),
+                dtype=torch.int64,
+                axis="tokens",
+            )
+        )
+        r.register_slot(
+            GraphSlot(
+                name="req_pool_indices",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int64,
+                axis="bs",
+                padding_policy=PaddingPolicy.ZERO,
+            )
+        )
+        r.register_slot(
+            GraphSlot(
+                name="seq_lens",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+                padding_policy=PaddingPolicy.FILL_SENTINEL,
+                pad_value=5,
+            )
+        )
+        r.register_slot(
+            GraphSlot(
+                name="out_cache_loc",
+                shape_fn=lambda bs, mt: (mt,),
+                dtype=torch.int64,
+                axis="tokens",
+                padding_policy=PaddingPolicy.ZERO,
+            )
+        )
+        r.register_slot(
+            GraphSlot(
+                name="positions",
+                shape_fn=lambda bs, mt: (mt,),
+                dtype=torch.int64,
+                axis="tokens",
+            )
+        )
+        r.register_slot(
+            GraphSlot(
+                name="seq_lens_cpu",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+                device=torch.device("cpu"),
+                padding_policy=PaddingPolicy.FILL_SENTINEL,
+                pad_value=5,
+            )
+        )
+        return r
+
+    def test_basic_fill_no_padding(self):
+        r = self._build_registry()
+        fb = _MiniForwardBatch(
+            batch_size=4,
+            input_ids=torch.arange(8, dtype=torch.int64),
+            req_pool_indices=torch.tensor([3, 1, 4, 2], dtype=torch.int64),
+            seq_lens=torch.tensor([10, 11, 12, 13], dtype=torch.int32),
+            out_cache_loc=torch.arange(8, dtype=torch.int64) + 100,
+            positions=torch.arange(8, dtype=torch.int64),
+            seq_lens_cpu=torch.tensor([10, 11, 12, 13], dtype=torch.int32),
+        )
+        r.fill_from(
+            fb,
+            raw_bs=4,
+            padded_bs=4,
+            raw_num_tokens=8,
+            padded_num_tokens=8,
+        )
+        self.assertTrue(torch.equal(r.get_slot("input_ids").buffer, fb.input_ids))
+        self.assertTrue(
+            torch.equal(r.get_slot("req_pool_indices").buffer, fb.req_pool_indices)
+        )
+        self.assertTrue(torch.equal(r.get_slot("seq_lens").buffer, fb.seq_lens))
+        self.assertTrue(
+            torch.equal(r.get_slot("out_cache_loc").buffer, fb.out_cache_loc)
+        )
+        self.assertTrue(torch.equal(r.get_slot("positions").buffer, fb.positions))
+        self.assertTrue(torch.equal(r.get_slot("seq_lens_cpu").buffer, fb.seq_lens_cpu))
+
+    def test_fill_with_padding_resets_zero_and_sentinel(self):
+        r = self._build_registry()
+        # Pre-poison the padded tail to a non-zero value so we can prove
+        # the reset_padding step ran.
+        r.get_slot("req_pool_indices").buffer.fill_(99)
+        r.get_slot("seq_lens").buffer.fill_(99)
+        r.get_slot("out_cache_loc").buffer.fill_(99)
+        # Raw 2 reqs, 4 tokens; padded 4 reqs, 8 tokens.
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            input_ids=torch.arange(4, dtype=torch.int64),
+            req_pool_indices=torch.tensor([3, 1], dtype=torch.int64),
+            seq_lens=torch.tensor([10, 11], dtype=torch.int32),
+            out_cache_loc=torch.arange(4, dtype=torch.int64) + 100,
+            positions=torch.arange(4, dtype=torch.int64),
+            seq_lens_cpu=torch.tensor([10, 11], dtype=torch.int32),
+        )
+        r.fill_from(
+            fb,
+            raw_bs=2,
+            padded_bs=4,
+            raw_num_tokens=4,
+            padded_num_tokens=8,
+        )
+        # Raw region copied.
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("req_pool_indices").buffer[:2],
+                torch.tensor([3, 1], dtype=torch.int64),
+            )
+        )
+        # ZERO padding tail.
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("req_pool_indices").buffer[2:],
+                torch.zeros(2, dtype=torch.int64),
+            )
+        )
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("out_cache_loc").buffer[4:],
+                torch.zeros(4, dtype=torch.int64),
+            )
+        )
+        # FILL_SENTINEL padding tail.
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("seq_lens").buffer[2:],
+                torch.tensor([5, 5], dtype=torch.int32),
+            )
+        )
+        # seq_lens_cpu lives on CPU device.
+        self.assertEqual(r.get_slot("seq_lens_cpu").buffer.device.type, "cpu")
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("seq_lens_cpu").buffer[2:],
+                torch.tensor([5, 5], dtype=torch.int32),
+            )
+        )
+
+    def test_keep_pad_preserves_padded_tail(self):
+        r = _make_registry(max_bs=4, max_num_tokens=8)
+        r.register_slot(
+            GraphSlot(
+                name="positions",
+                shape_fn=lambda bs, mt: (mt,),
+                dtype=torch.int64,
+                axis="tokens",
+                padding_policy=PaddingPolicy.KEEP_PAD,
+            )
+        )
+        # Poison padded tail.
+        r.get_slot("positions").buffer.fill_(77)
+        fb = _MiniForwardBatch(
+            batch_size=1,
+            positions=torch.arange(2, dtype=torch.int64),
+        )
+        r.fill_from(
+            fb,
+            raw_bs=1,
+            padded_bs=4,
+            raw_num_tokens=2,
+            padded_num_tokens=8,
+        )
+        # Raw is copied; padded tail stays at the poison value.
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("positions").buffer[:2],
+                torch.tensor([0, 1], dtype=torch.int64),
+            )
+        )
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("positions").buffer[2:],
+                torch.full((6,), 77, dtype=torch.int64),
+            )
+        )
+
+    def test_extract_buffer_returns_fb_view(self):
+        r = self._build_registry()
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            input_ids=torch.arange(4, dtype=torch.int64),
+            req_pool_indices=torch.tensor([3, 1], dtype=torch.int64),
+            seq_lens=torch.tensor([10, 11], dtype=torch.int32),
+            out_cache_loc=torch.arange(4, dtype=torch.int64) + 100,
+            positions=torch.arange(4, dtype=torch.int64),
+            seq_lens_cpu=torch.tensor([10, 11], dtype=torch.int32),
+            forward_mode="DECODE",
+            spec_info="dummy_spec",
+        )
+        r.fill_from(
+            fb,
+            raw_bs=2,
+            padded_bs=4,
+            raw_num_tokens=4,
+            padded_num_tokens=8,
+        )
+        fb_view = r.extract_buffer(
+            padded_bs=4,
+            padded_num_tokens=8,
+            forward_batch_template=fb,
+        )
+        # batch_size is padded.
+        self.assertEqual(fb_view.batch_size, 4)
+        # Tensor fields are now buffer views — same data_ptr as the slot.
+        self.assertEqual(
+            fb_view.input_ids.data_ptr(),
+            r.get_slot("input_ids").buffer.data_ptr(),
+        )
+        # Length is padded.
+        self.assertEqual(len(fb_view.input_ids), 8)
+        self.assertEqual(len(fb_view.req_pool_indices), 4)
+        # Non-slot fields carried from template.
+        self.assertEqual(fb_view.forward_mode, "DECODE")
+        self.assertEqual(fb_view.spec_info, "dummy_spec")
+        # Template itself NOT mutated (replace returns a new instance).
+        self.assertEqual(fb.batch_size, 2)
+        self.assertIsNot(fb_view, fb)
+
+
+class TestMissingAndOptionalSlots(unittest.TestCase):
+    def test_missing_fb_attr_is_skipped(self):
+        r = _make_registry()
+        r.register_slot(
+            GraphSlot(
+                name="encoder_lens",
+                shape_fn=lambda bs, mt: (bs,),
+                dtype=torch.int32,
+                axis="bs",
+                padding_policy=PaddingPolicy.FILL_SENTINEL,
+                pad_value=0,
+            )
+        )
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            input_ids=torch.arange(4, dtype=torch.int64),
+            encoder_lens=None,  # FB doesn't carry this for this request.
+        )
+        # Should NOT raise; encoder_lens buffer stays at the FILL_SENTINEL
+        # init value.
+        r.fill_from(
+            fb,
+            raw_bs=2,
+            padded_bs=4,
+            raw_num_tokens=4,
+            padded_num_tokens=8,
+        )
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("encoder_lens").buffer,
+                torch.zeros(8, dtype=torch.int32),
+            )
+        )
+
+
+class TestPostFillHook(unittest.TestCase):
+    def test_post_fill_runs_after_copy(self):
+        observed = {}
+
+        def hook(buf, fb, ctx):
+            # Multiply the raw region by 10 in-place.
+            buf[: ctx.raw_num_tokens] *= 10
+            observed["raw_n"] = ctx.raw_num_tokens
+            observed["padded_n"] = ctx.padded_num_tokens
+
+        r = _make_registry(max_bs=4, max_num_tokens=8)
+        r.register_slot(
+            GraphSlot(
+                name="input_ids",
+                shape_fn=lambda bs, mt: (mt,),
+                dtype=torch.int64,
+                axis="tokens",
+                post_fill=hook,
+            )
+        )
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            input_ids=torch.arange(1, 5, dtype=torch.int64),  # [1,2,3,4]
+        )
+        r.fill_from(
+            fb,
+            raw_bs=2,
+            padded_bs=4,
+            raw_num_tokens=4,
+            padded_num_tokens=8,
+        )
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("input_ids").buffer[:4],
+                torch.tensor([10, 20, 30, 40], dtype=torch.int64),
+            )
+        )
+        self.assertEqual(observed, {"raw_n": 4, "padded_n": 8})
+
+
+class TestSliceFnSlot(unittest.TestCase):
+    """``mrope_positions`` has shape ``[3, T]`` — sliced on axis 1."""
+
+    def test_slice_fn_handles_2d_tokens_axis(self):
+        r = _make_registry(max_bs=4, max_num_tokens=8)
+        r.register_slot(
+            GraphSlot(
+                name="mrope_positions",
+                shape_fn=lambda bs, mt: (3, mt),
+                dtype=torch.int64,
+                axis="tokens",
+                slice_fn=lambda buf, n: buf[:, :n],
+            )
+        )
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            mrope_positions=torch.arange(12, dtype=torch.int64).reshape(3, 4),
+        )
+        r.fill_from(
+            fb,
+            raw_bs=2,
+            padded_bs=4,
+            raw_num_tokens=4,
+            padded_num_tokens=8,
+        )
+        # Raw 3x4 region copied.
+        self.assertTrue(
+            torch.equal(
+                r.get_slot("mrope_positions").buffer[:, :4],
+                fb.mrope_positions,
+            )
+        )
+        # extract_buffer should hand back the [:, :padded_num_tokens] view.
+        fb_view = r.extract_buffer(
+            padded_bs=4,
+            padded_num_tokens=8,
+            forward_batch_template=fb,
+        )
+        self.assertEqual(fb_view.mrope_positions.shape, (3, 8))
+
+
+class TestSourceFnSlots(unittest.TestCase):
+    """``source_fn`` slots copy from a nested FB field or a side input, with a
+    source-length slice, and are skipped by ``extract_buffer``."""
+
+    def test_nested_fb_source_copies_source_length_head(self):
+        r = _make_registry(max_bs=8, max_num_tokens=16)
+        r.register_slot(
+            GraphSlot(
+                name="ngram_embedding_info.column_starts",
+                shape_fn=lambda _bs, _mt: (8,),
+                dtype=torch.int32,
+                axis="none",
+                padding_policy=PaddingPolicy.KEEP_PAD,
+                source_fn=lambda fb, ctx: (
+                    None
+                    if fb.ngram_embedding_info is None
+                    else fb.ngram_embedding_info.column_starts
+                ),
+            )
+        )
+        buf = r.get_slot("ngram_embedding_info.column_starts").buffer
+        buf.fill_(99)  # sentinel to prove the tail is untouched
+        fb = _MiniForwardBatch(
+            batch_size=3,
+            ngram_embedding_info=SimpleNamespace(
+                column_starts=torch.tensor([1, 2, 3], dtype=torch.int32),
+            ),
+        )
+        r.fill_from(fb, raw_bs=3, padded_bs=8, raw_num_tokens=3, padded_num_tokens=16)
+        # Head [:3] copied from the source; tail [3:] kept as the sentinel.
+        self.assertTrue(
+            torch.equal(buf[:3], torch.tensor([1, 2, 3], dtype=torch.int32))
+        )
+        self.assertTrue(torch.all(buf[3:] == 99))
+
+    def test_source_fn_returning_none_skips_copy(self):
+        r = _make_registry(max_bs=8, max_num_tokens=16)
+        r.register_slot(
+            GraphSlot(
+                name="ngram_embedding_info.column_starts",
+                shape_fn=lambda _bs, _mt: (8,),
+                dtype=torch.int32,
+                axis="none",
+                padding_policy=PaddingPolicy.KEEP_PAD,
+                source_fn=lambda fb, ctx: (
+                    None
+                    if fb.ngram_embedding_info is None
+                    else fb.ngram_embedding_info.column_starts
+                ),
+            )
+        )
+        buf = r.get_slot("ngram_embedding_info.column_starts").buffer
+        buf.fill_(7)
+        fb = _MiniForwardBatch(batch_size=3, ngram_embedding_info=None)
+        r.fill_from(fb, raw_bs=3, padded_bs=8, raw_num_tokens=3, padded_num_tokens=16)
+        self.assertTrue(torch.all(buf == 7))  # untouched
+
+    def test_side_input_source_via_fill_context(self):
+        r = _make_registry(max_bs=8, max_num_tokens=16)
+        r.register_slot(
+            GraphSlot(
+                name="pp_proxy_tensors.hidden_states",
+                shape_fn=lambda _bs, mt: (mt,),
+                dtype=torch.int32,
+                axis="none",
+                padding_policy=PaddingPolicy.KEEP_PAD,
+                source_fn=lambda fb, ctx: (
+                    None
+                    if ctx.pp_proxy_tensors is None
+                    else ctx.pp_proxy_tensors.tensors["hidden_states"]
+                ),
+            )
+        )
+        buf = r.get_slot("pp_proxy_tensors.hidden_states").buffer
+        buf.zero_()
+        fb = _MiniForwardBatch(batch_size=4)
+        pp = SimpleNamespace(
+            tensors={"hidden_states": torch.tensor([5, 6, 7, 8], dtype=torch.int32)}
+        )
+        r.fill_from(
+            fb,
+            raw_bs=4,
+            padded_bs=8,
+            raw_num_tokens=4,
+            padded_num_tokens=16,
+            pp_proxy_tensors=pp,
+        )
+        self.assertTrue(
+            torch.equal(buf[:4], torch.tensor([5, 6, 7, 8], dtype=torch.int32))
+        )
+
+    def test_extract_buffer_skips_dotted_slots(self):
+        r = _make_registry(max_bs=8, max_num_tokens=16)
+        r.register_slot(
+            GraphSlot(
+                name="ngram_embedding_info.column_starts",
+                shape_fn=lambda _bs, _mt: (8,),
+                dtype=torch.int32,
+                axis="none",
+                padding_policy=PaddingPolicy.KEEP_PAD,
+                source_fn=lambda fb, ctx: None,
+            )
+        )
+        fb = _MiniForwardBatch(batch_size=3)
+        # dataclasses.replace must not be handed a dotted kwarg.
+        fb_view = r.extract_buffer(
+            padded_bs=8, padded_num_tokens=16, forward_batch_template=fb
+        )
+        self.assertEqual(fb_view.batch_size, 8)
+
+
+class TestPoolBackedAlloc(unittest.TestCase):
+    """``share_pool=True`` coalesces same-named slot buffers through the
+    global ForwardInputBuffers pool (so a registry can share storage with
+    the legacy DecodeInputBuffers during migration)."""
+
+    def setUp(self):
+        from sglang.srt.model_executor import input_buffers
+
+        input_buffers._forward_input_buffer_pool.clear()
+
+    def _reg(self, *, max_num_tokens=16, share_pool):
+        return CudaGraphBufferRegistry(
+            device=torch.device("cpu"),
+            max_bs=8,
+            max_num_tokens=max_num_tokens,
+            share_pool=share_pool,
+        )
+
+    @staticmethod
+    def _ids_slot(name):
+        return GraphSlot(
+            name=name,
+            shape_fn=lambda bs, mt: (mt,),
+            dtype=torch.int64,
+            axis="tokens",
+        )
+
+    def test_share_pool_off_is_independent(self):
+        r1, r2 = self._reg(share_pool=False), self._reg(share_pool=False)
+        r1.register_slot(self._ids_slot("ids"))
+        r2.register_slot(self._ids_slot("ids"))
+        self.assertNotEqual(
+            r1.get_slot("ids").buffer.data_ptr(),
+            r2.get_slot("ids").buffer.data_ptr(),
+        )
+
+    def test_same_size_shares_one_allocation(self):
+        a = self._reg(max_num_tokens=16, share_pool=True)
+        b = self._reg(max_num_tokens=16, share_pool=True)
+        a.register_slot(self._ids_slot("ids"))
+        b.register_slot(self._ids_slot("ids"))
+        # Identical (name, size, dtype, device) -> one shared allocation.
+        self.assertEqual(
+            a.get_slot("ids").buffer.data_ptr(),
+            b.get_slot("ids").buffer.data_ptr(),
+        )
+
+    def test_different_sizes_do_not_share(self):
+        big = self._reg(max_num_tokens=32, share_pool=True)
+        small = self._reg(max_num_tokens=16, share_pool=True)
+        big.register_slot(self._ids_slot("ids"))
+        small.register_slot(self._ids_slot("ids"))
+        # Different sizes -> different pool keys -> independent storage (no
+        # aliasing a smaller request onto a larger buffer).
+        self.assertEqual(tuple(small.get_slot("ids").buffer.shape), (16,))
+        self.assertEqual(tuple(big.get_slot("ids").buffer.shape), (32,))
+        self.assertNotEqual(
+            small.get_slot("ids").buffer.data_ptr(),
+            big.get_slot("ids").buffer.data_ptr(),
+        )
+
+    def test_sharing_is_independent_of_registration_order(self):
+        from sglang.srt.model_executor import input_buffers
+
+        def _ptrs(first_tokens, second_tokens):
+            input_buffers._forward_input_buffer_pool.clear()
+            r1 = self._reg(max_num_tokens=first_tokens, share_pool=True)
+            r1.register_slot(self._ids_slot("ids"))
+            r2 = self._reg(max_num_tokens=second_tokens, share_pool=True)
+            r2.register_slot(self._ids_slot("ids"))
+            return r1.get_slot("ids").buffer, r2.get_slot("ids").buffer
+
+        # Same size: shares in either order.
+        a, b = _ptrs(16, 16)
+        self.assertEqual(a.data_ptr(), b.data_ptr())
+        b2, a2 = _ptrs(16, 16)
+        self.assertEqual(a2.data_ptr(), b2.data_ptr())
+
+        # Different sizes: never shares, regardless of which registers first
+        # (the old strictly-larger rule shared big-then-small but not
+        # small-then-big — that asymmetry is what this fix removes).
+        big_first, small_after = _ptrs(32, 16)
+        self.assertNotEqual(big_first.data_ptr(), small_after.data_ptr())
+        small_first, big_after = _ptrs(16, 32)
+        self.assertNotEqual(small_first.data_ptr(), big_after.data_ptr())
+
+
+class TestBuildDecodeRegistry(unittest.TestCase):
+    """``build_decode_registry`` registers the always-on FB-shared decode
+    slots with padding policies matching
+    ``DecodeInputBuffers.populate_from_forward_batch``."""
+
+    def setUp(self):
+        from sglang.srt.model_executor import input_buffers
+
+        input_buffers._forward_input_buffer_pool.clear()
+
+    def test_factory_slot_set_and_padding(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        FILL = 5
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=FILL,
+            cache_loc_dtype=torch.int64,
+            share_pool=False,
+        )
+        for name in (
+            "input_ids",
+            "positions",
+            "out_cache_loc",
+            "req_pool_indices",
+            "seq_lens",
+            "seq_lens_cpu",
+            "mrope_positions",
+        ):
+            self.assertTrue(reg.has_slot(name), name)
+        self.assertFalse(reg.has_slot("mamba_track_indices"))
+
+        raw_bs, padded_bs, raw_nt, padded_nt = 2, 4, 2, 4
+        fb = _MiniForwardBatch(
+            batch_size=raw_bs,
+            input_ids=torch.tensor([10, 11], dtype=torch.int64),
+            positions=torch.tensor([0, 1], dtype=torch.int64),
+            out_cache_loc=torch.tensor([100, 101], dtype=torch.int64),
+            req_pool_indices=torch.tensor([1, 2], dtype=torch.int64),
+            seq_lens=torch.tensor([7, 8], dtype=torch.int32),
+            seq_lens_cpu=torch.tensor([7, 8], dtype=torch.int32),
+            mrope_positions=torch.tensor([[0, 1], [0, 1], [0, 1]], dtype=torch.int64),
+        )
+        # Poison tails so resets are observable.
+        for n in ("input_ids", "positions", "out_cache_loc", "req_pool_indices"):
+            reg.get_slot(n).buffer.fill_(99)
+        reg.fill_from(
+            fb,
+            raw_bs=raw_bs,
+            padded_bs=padded_bs,
+            raw_num_tokens=raw_nt,
+            padded_num_tokens=padded_nt,
+        )
+
+        # FOREACH_COPY: head copied, tail kept (poison).
+        ids = reg.get_slot("input_ids").buffer
+        self.assertTrue(torch.equal(ids[:2], torch.tensor([10, 11])))
+        self.assertTrue(torch.equal(ids[2:4], torch.tensor([99, 99])))
+        # ZERO: head copied, tail zeroed.
+        oc = reg.get_slot("out_cache_loc").buffer
+        self.assertTrue(torch.equal(oc[:2], torch.tensor([100, 101])))
+        self.assertTrue(torch.equal(oc[2:4], torch.tensor([0, 0])))
+        rp = reg.get_slot("req_pool_indices").buffer
+        self.assertTrue(torch.equal(rp[:2], torch.tensor([1, 2])))
+        self.assertTrue(torch.equal(rp[2:4], torch.tensor([0, 0])))
+        # FILL_SENTINEL: head copied, tail = seq_len_fill_value.
+        sl = reg.get_slot("seq_lens").buffer
+        self.assertTrue(torch.equal(sl[:2], torch.tensor([7, 8], dtype=torch.int32)))
+        self.assertTrue(
+            torch.equal(sl[2:4], torch.tensor([FILL, FILL], dtype=torch.int32))
+        )
+        slc = reg.get_slot("seq_lens_cpu").buffer
+        self.assertEqual(slc.device.type, "cpu")
+        self.assertTrue(
+            torch.equal(slc[2:4], torch.tensor([FILL, FILL], dtype=torch.int32))
+        )
+        # 2D mrope via slice_fn.
+        mr = reg.get_slot("mrope_positions").buffer
+        self.assertTrue(torch.equal(mr[:, :2], fb.mrope_positions))
+
+        fb_view = reg.extract_buffer(
+            padded_bs=padded_bs,
+            padded_num_tokens=padded_nt,
+            forward_batch_template=fb,
+        )
+        self.assertEqual(fb_view.batch_size, padded_bs)
+        self.assertEqual(fb_view.input_ids.shape[0], padded_nt)
+        self.assertEqual(fb_view.seq_lens.shape[0], padded_bs)
+
+    def test_source_adopts_buffers(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        src = SimpleNamespace(
+            input_ids=torch.zeros(8, dtype=torch.int64),
+            positions=torch.zeros(8, dtype=torch.int64),
+            out_cache_loc=torch.zeros(8, dtype=torch.int64),
+            req_pool_indices=torch.zeros(4, dtype=torch.int64),
+            seq_lens=torch.full((4,), 5, dtype=torch.int32),
+            seq_lens_cpu=torch.full((4,), 5, dtype=torch.int32),
+            mrope_positions=torch.zeros((3, 8), dtype=torch.int64),
+            global_num_tokens_gpu=torch.zeros(1, dtype=torch.int32),
+            global_num_tokens_for_logprob_gpu=torch.zeros(1, dtype=torch.int32),
+        )
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=5,
+            cache_loc_dtype=torch.int64,
+            source=src,
+        )
+        # Registry slots share storage with the source's tensors.
+        for name in ("input_ids", "seq_lens", "seq_lens_cpu", "mrope_positions"):
+            self.assertEqual(
+                reg.get_slot(name).buffer.data_ptr(),
+                getattr(src, name).data_ptr(),
+                name,
+            )
+
+    def test_source_with_ngram_registers_structured_slots(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        col = torch.zeros(4, dtype=torch.int32)
+        req = torch.ones(4, dtype=torch.int32)
+        src = SimpleNamespace(
+            input_ids=torch.zeros(8, dtype=torch.int64),
+            positions=torch.zeros(8, dtype=torch.int64),
+            out_cache_loc=torch.zeros(8, dtype=torch.int64),
+            req_pool_indices=torch.zeros(4, dtype=torch.int64),
+            seq_lens=torch.full((4,), 5, dtype=torch.int32),
+            seq_lens_cpu=torch.full((4,), 5, dtype=torch.int32),
+            mrope_positions=torch.zeros((3, 8), dtype=torch.int64),
+            global_num_tokens_gpu=torch.zeros(1, dtype=torch.int32),
+            global_num_tokens_for_logprob_gpu=torch.zeros(1, dtype=torch.int32),
+            ngram_embedding_info=SimpleNamespace(column_starts=col, req_lens=req),
+        )
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=5,
+            cache_loc_dtype=torch.int64,
+            source=src,
+        )
+        # Structured slots adopt the source's nested storage.
+        self.assertTrue(reg.has_slot("ngram_embedding_info.column_starts"))
+        self.assertEqual(
+            reg.get_slot("ngram_embedding_info.column_starts").buffer.data_ptr(),
+            col.data_ptr(),
+        )
+        # And fill_from copies the head from the FB's nested dataclass.
+        fb = _MiniForwardBatch(
+            batch_size=3,
+            ngram_embedding_info=SimpleNamespace(
+                column_starts=torch.tensor([7, 8, 9], dtype=torch.int32),
+                req_lens=torch.tensor([1, 1, 2], dtype=torch.int32),
+            ),
+        )
+        reg.fill_from(fb, raw_bs=3, padded_bs=4, raw_num_tokens=3, padded_num_tokens=8)
+        self.assertTrue(
+            torch.equal(col[:3], torch.tensor([7, 8, 9], dtype=torch.int32))
+        )
+        self.assertTrue(
+            torch.equal(req[:3], torch.tensor([1, 1, 2], dtype=torch.int32))
+        )
+
+    def test_source_with_pp_registers_proxy_slots(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        hs = torch.zeros((8, 2), dtype=torch.int32)
+        src = SimpleNamespace(
+            input_ids=torch.zeros(8, dtype=torch.int64),
+            positions=torch.zeros(8, dtype=torch.int64),
+            out_cache_loc=torch.zeros(8, dtype=torch.int64),
+            req_pool_indices=torch.zeros(4, dtype=torch.int64),
+            seq_lens=torch.full((4,), 5, dtype=torch.int32),
+            seq_lens_cpu=torch.full((4,), 5, dtype=torch.int32),
+            mrope_positions=torch.zeros((3, 8), dtype=torch.int64),
+            global_num_tokens_gpu=torch.zeros(1, dtype=torch.int32),
+            global_num_tokens_for_logprob_gpu=torch.zeros(1, dtype=torch.int32),
+            pp_proxy_tensors={"hidden_states": hs},
+        )
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=5,
+            cache_loc_dtype=torch.int64,
+            source=src,
+        )
+        self.assertTrue(reg.has_slot("pp_proxy_tensors.hidden_states"))
+        self.assertEqual(
+            reg.get_slot("pp_proxy_tensors.hidden_states").buffer.data_ptr(),
+            hs.data_ptr(),
+        )
+        # The pp input is not on the FB — it rides on the fill_from kwarg.
+        fb = _MiniForwardBatch(batch_size=3)
+        pp = SimpleNamespace(
+            tensors={"hidden_states": torch.ones((3, 2), dtype=torch.int32)}
+        )
+        reg.fill_from(
+            fb,
+            raw_bs=3,
+            padded_bs=4,
+            raw_num_tokens=3,
+            padded_num_tokens=8,
+            pp_proxy_tensors=pp,
+        )
+        self.assertTrue(torch.all(hs[:3] == 1))
+        self.assertTrue(torch.all(hs[3:] == 0))  # tail untouched
+
+    def test_source_with_canary_registers_bs_slots(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        rids = torch.zeros(4, dtype=torch.int64)
+        boot = torch.full((4,), -1, dtype=torch.int64)
+        src = SimpleNamespace(
+            input_ids=torch.zeros(8, dtype=torch.int64),
+            positions=torch.zeros(8, dtype=torch.int64),
+            out_cache_loc=torch.zeros(8, dtype=torch.int64),
+            req_pool_indices=torch.zeros(4, dtype=torch.int64),
+            seq_lens=torch.full((4,), 5, dtype=torch.int32),
+            seq_lens_cpu=torch.full((4,), 5, dtype=torch.int32),
+            mrope_positions=torch.zeros((3, 8), dtype=torch.int64),
+            global_num_tokens_gpu=torch.zeros(1, dtype=torch.int32),
+            global_num_tokens_for_logprob_gpu=torch.zeros(1, dtype=torch.int32),
+            rids_int=rids,
+            bootstrap_room_ids_int=boot,
+        )
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=5,
+            cache_loc_dtype=torch.int64,
+            source=src,
+        )
+        self.assertTrue(reg.has_slot("rids_int"))
+        self.assertTrue(reg.has_slot("bootstrap_room_ids_int"))
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            rids_int=torch.tensor([10, 11], dtype=torch.int64),
+            bootstrap_room_ids_int=torch.tensor([20, 21], dtype=torch.int64),
+        )
+        reg.fill_from(fb, raw_bs=2, padded_bs=4, raw_num_tokens=2, padded_num_tokens=8)
+        # Head copied; bootstrap tail keeps its -1 init (no per-iter reset).
+        self.assertTrue(
+            torch.equal(rids[:2], torch.tensor([10, 11], dtype=torch.int64))
+        )
+        self.assertTrue(
+            torch.equal(boot[:2], torch.tensor([20, 21], dtype=torch.int64))
+        )
+        self.assertTrue(torch.all(boot[2:] == -1))
+
+
+class TestBuildPrefillRegistry(unittest.TestCase):
+    """Token-axis prefill registry (piecewise / breakable runners): ZERO-tail
+    padding, input_embeds reset-only, mamba bs-axis copy, source adoption."""
+
+    def _src(self, **extra):
+        base = dict(
+            input_ids=torch.zeros(16, dtype=torch.int64),
+            positions=torch.zeros(16, dtype=torch.int64),
+            out_cache_loc=torch.zeros(16, dtype=torch.int64),
+        )
+        base.update(extra)
+        return SimpleNamespace(**base)
+
+    def test_core_token_slots_zero_tail_and_copy_head(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_prefill_registry,
+        )
+
+        src = self._src()
+        reg = build_prefill_registry(
+            device=torch.device("cpu"),
+            max_bs=1,
+            max_num_token=16,
+            cache_loc_dtype=torch.int64,
+            source=src,
+        )
+        for name in ("input_ids", "positions", "out_cache_loc"):
+            self.assertEqual(
+                reg.get_slot(name).buffer.data_ptr(),
+                getattr(src, name).data_ptr(),
+                name,
+            )
+        # poison tails so the ZERO reset is observable
+        for name in ("input_ids", "positions", "out_cache_loc"):
+            reg.get_slot(name).buffer.fill_(7)
+        fb = _MiniForwardBatch(
+            input_ids=torch.tensor([1, 2, 3], dtype=torch.int64),
+            positions=torch.tensor([4, 5, 6], dtype=torch.int64),
+            out_cache_loc=torch.tensor([8, 9, 10], dtype=torch.int64),
+        )
+        # raw 3 tokens, padded (static) bucket 8
+        reg.fill_from(fb, raw_bs=1, padded_bs=1, raw_num_tokens=3, padded_num_tokens=8)
+        ids = reg.get_slot("input_ids").buffer
+        self.assertTrue(
+            torch.equal(ids[:3], torch.tensor([1, 2, 3], dtype=torch.int64))
+        )
+        self.assertTrue(torch.all(ids[3:8] == 0))  # padded tail reset
+        self.assertTrue(torch.all(ids[8:] == 7))  # beyond the bucket: untouched
+
+    def test_multimodal_input_embeds_reset_only(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_prefill_registry,
+        )
+
+        src = self._src(
+            mrope_positions=torch.zeros((3, 16), dtype=torch.int64),
+            input_embeds=torch.zeros((16, 4), dtype=torch.float32),
+        )
+        reg = build_prefill_registry(
+            device=torch.device("cpu"),
+            max_bs=1,
+            max_num_token=16,
+            cache_loc_dtype=torch.int64,
+            is_multimodal=True,
+            hidden_size=4,
+            embed_dtype=torch.float32,
+            source=src,
+        )
+        self.assertTrue(reg.has_slot("mrope_positions"))
+        self.assertTrue(reg.has_slot("input_embeds"))
+        emb = reg.get_slot("input_embeds").buffer
+        emb.fill_(5.0)  # the model would write real embeds here; we just check reset
+        fb = _MiniForwardBatch(
+            input_ids=torch.zeros(3, dtype=torch.int64),
+            positions=torch.zeros(3, dtype=torch.int64),
+            out_cache_loc=torch.zeros(3, dtype=torch.int64),
+            mrope_positions=torch.ones((3, 3), dtype=torch.int64),
+            input_embeds=torch.full(
+                (3, 4), 9.0
+            ),  # must be ignored (copy_from_fb=False)
+        )
+        reg.fill_from(fb, raw_bs=1, padded_bs=1, raw_num_tokens=3, padded_num_tokens=8)
+        # input_embeds: head NOT copied from FB; padded tail zeroed.
+        self.assertTrue(torch.all(emb[:3] == 5.0))
+        self.assertTrue(torch.all(emb[3:8] == 0.0))
+        # mrope: head copied, 2D tail zeroed.
+        mr = reg.get_slot("mrope_positions").buffer
+        self.assertTrue(torch.all(mr[:, :3] == 1))
+        self.assertTrue(torch.all(mr[:, 3:8] == 0))
+
+    def test_mamba_bs_axis_copy(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_prefill_registry,
+        )
+
+        idx = torch.zeros(2, dtype=torch.int64)
+        src = self._src(
+            mamba_track_indices=idx,
+            mamba_track_mask=torch.zeros(2, dtype=torch.bool),
+            mamba_track_seqlens=torch.zeros(2, dtype=torch.int32),
+        )
+        reg = build_prefill_registry(
+            device=torch.device("cpu"),
+            max_bs=2,
+            max_num_token=16,
+            cache_loc_dtype=torch.int64,
+            enable_mamba_track=True,
+            source=src,
+        )
+        self.assertTrue(reg.has_slot("mamba_track_indices"))
+        fb = _MiniForwardBatch(
+            input_ids=torch.zeros(3, dtype=torch.int64),
+            positions=torch.zeros(3, dtype=torch.int64),
+            out_cache_loc=torch.zeros(3, dtype=torch.int64),
+            mamba_track_indices=torch.tensor([3, 4], dtype=torch.int64),
+            mamba_track_mask=torch.zeros(2, dtype=torch.bool),
+            mamba_track_seqlens=torch.zeros(2, dtype=torch.int32),
+        )
+        reg.fill_from(fb, raw_bs=2, padded_bs=2, raw_num_tokens=3, padded_num_tokens=8)
+        self.assertTrue(torch.equal(idx, torch.tensor([3, 4], dtype=torch.int64)))
+
+
+class TestFillOncePolicy(unittest.TestCase):
+    """FILL_ONCE initializes the whole buffer at alloc and never resets the
+    padded tail per iter (unlike FILL_SENTINEL)."""
+
+    def test_fill_once_inits_once_and_keeps_tail(self):
+        reg = CudaGraphBufferRegistry(
+            device=torch.device("cpu"), max_bs=4, max_num_tokens=8
+        )
+        reg.register_slot(
+            GraphSlot(
+                "encoder_lens",
+                lambda bs, mt: (bs,),
+                torch.int32,
+                axis="bs",
+                padding_policy=PaddingPolicy.FILL_ONCE,
+                pad_value=9,
+            )
+        )
+        buf = reg.get_slot("encoder_lens").buffer
+        self.assertTrue(torch.equal(buf, torch.tensor([9, 9, 9, 9], dtype=torch.int32)))
+        buf[2:].fill_(99)  # poison the tail
+        fb = _MiniForwardBatch(
+            batch_size=2, encoder_lens=torch.tensor([1, 2], dtype=torch.int32)
+        )
+        reg.fill_from(fb, raw_bs=2, padded_bs=4, raw_num_tokens=2, padded_num_tokens=4)
+        # Head copied; tail NOT reset (FILL_ONCE skips the per-iter reset).
+        self.assertTrue(torch.equal(buf[:2], torch.tensor([1, 2], dtype=torch.int32)))
+        self.assertTrue(torch.equal(buf[2:], torch.tensor([99, 99], dtype=torch.int32)))
+
+
+class TestComputedSlots(unittest.TestCase):
+    """num_token_non_padded (copy_from_fb + post_fill) and global_num_tokens
+    (copy_from_fb=False + post_fill fill)."""
+
+    def test_num_token_non_padded_copy_path(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=5,
+            cache_loc_dtype=torch.int64,
+            enable_num_token_non_padded=True,
+            require_gathered_buffer=False,
+        )
+        self.assertTrue(reg.has_slot("num_token_non_padded"))
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            num_token_non_padded=torch.tensor([7], dtype=torch.int32),
+        )
+        reg.fill_from(fb, raw_bs=2, padded_bs=2, raw_num_tokens=2, padded_num_tokens=2)
+        # Non-gathered: plain FB copy, post_fill is a no-op.
+        self.assertTrue(
+            torch.equal(
+                reg.get_slot("num_token_non_padded").buffer,
+                torch.tensor([7], dtype=torch.int32),
+            )
+        )
+
+    def test_global_num_tokens_fill_path(self):
+        from sglang.srt.model_executor.cuda_graph_buffer_registry import (
+            build_decode_registry,
+        )
+
+        reg = build_decode_registry(
+            device=torch.device("cpu"),
+            max_bs=4,
+            max_num_token=8,
+            seq_len_fill_value=5,
+            cache_loc_dtype=torch.int64,
+            require_gathered_buffer=True,
+        )
+        self.assertTrue(reg.has_slot("global_num_tokens_gpu"))
+        # FB carries a stale value; copy_from_fb=False means it's ignored and
+        # the slot is filled with padded_num_tokens by post_fill.
+        fb = _MiniForwardBatch(
+            batch_size=2,
+            global_num_tokens_gpu=torch.tensor([999], dtype=torch.int32),
+        )
+        reg.fill_from(fb, raw_bs=2, padded_bs=4, raw_num_tokens=2, padded_num_tokens=4)
+        self.assertTrue(
+            torch.equal(
+                reg.get_slot("global_num_tokens_gpu").buffer,
+                torch.tensor([4], dtype=torch.int32),
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()