diff --git a/tensorrt_llm/_torch/disaggregation/native/peer.py b/tensorrt_llm/_torch/disaggregation/native/peer.py
index d8d37658195b..6b98a2772ab1 100644
--- a/tensorrt_llm/_torch/disaggregation/native/peer.py
+++ b/tensorrt_llm/_torch/disaggregation/native/peer.py
@@ -180,6 +180,14 @@ def _check_peer_compatible(self, peer_ri: RankInfo) -> bool:
             peer_ri.page_table,
         )
         AttentionPolicy.validate_peer_compatible(self_page_table, peer_ri.page_table)
+        if self._ri.num_draft_layers != peer_ri.num_draft_layers:
+            # The layer-count tolerance below would transfer only the target
+            # layers and leave the drafter's prompt KV unset on the receiver.
+            raise ValueError(
+                "PeerRegistrar: unified draft KV layer count differs "
+                f"(local={self._ri.num_draft_layers}, peer={peer_ri.num_draft_layers}); "
+                "both peers must use the same speculative decoding KV cache layout."
+            )
 
         self_layers = sum(self._ri.layer_num_per_pp)
         peer_layers = sum(peer_ri.layer_num_per_pp)
diff --git a/tensorrt_llm/_torch/disaggregation/native/rank_info.py b/tensorrt_llm/_torch/disaggregation/native/rank_info.py
index 87731d303204..358b1b576412 100644
--- a/tensorrt_llm/_torch/disaggregation/native/rank_info.py
+++ b/tensorrt_llm/_torch/disaggregation/native/rank_info.py
@@ -49,6 +49,9 @@ class RankInfo:
     aux_meta: Optional[AuxBufferMeta] = None
     page_table: Optional[KVCachePageTable] = None
     bounded_replay_on_generation: bool = False
+    # Layers a unified KV cache manager owns for the drafter (layer specs with
+    # a DRAFT owner). They ride the target's pages, so both peers must agree.
+    num_draft_layers: int = 0
 
     @property
     def tp_size_per_dp_group(self) -> int:
@@ -114,6 +117,9 @@ def from_kv_cache_manager(
             aux_meta=aux_buffer_meta,
             page_table=build_page_table_from_manager(kvm),
             bounded_replay_on_generation=getattr(kvm, "bounded_replay_on_generation", False),
+            num_draft_layers=(
+                len(kvm.draft_layer_ids) if getattr(kvm, "layer_specs", None) is not None else 0
+            ),
         )
 
     @classmethod
diff --git a/tensorrt_llm/_torch/pyexecutor/_util.py b/tensorrt_llm/_torch/pyexecutor/_util.py
index 70c008a4069f..933bb1c3e432 100644
--- a/tensorrt_llm/_torch/pyexecutor/_util.py
+++ b/tensorrt_llm/_torch/pyexecutor/_util.py
@@ -41,6 +41,7 @@
 from tensorrt_llm.logger import logger
 from tensorrt_llm.mapping import CpType, Mapping
 from tensorrt_llm.quantization import QuantAlgo
+from tensorrt_llm.runtime.kv_cache_manager_v2 import AttentionReusePolicy
 
 from ..attention.backends import get_sparse_attn_kv_cache_manager
 from ..attention.backends.sparse.params import \
@@ -66,7 +67,9 @@
 from .engine.lora import make_lora_model_config
 from .engine.multimodal import is_multimodal, use_mrope
 from .guided_decoder import GuidedDecoder
-from .kv_cache.kv_cache_manager_v2 import KVCacheManagerV2
+from .kv_cache.kv_cache_manager_v2 import (KVCacheLayerOwner, KVCacheLayerSpec,
+                                           KVCacheManagerV2, Role,
+                                           _resolve_v2_max_attention_window_vec)
 from .kv_cache.mamba_cache_manager import (BaseMambaCacheManager,
                                            CppMambaHybridCacheManager,
                                            MambaHybridCacheManagerV2,
@@ -981,6 +984,8 @@ def __init__(
         self._sparse_attention_config = sparse_attention_config
         self._tokens_per_block = tokens_per_block
         self._max_seq_len = max_seq_len
+        # Estimation can shrink _max_seq_len; cache ownership must stay fixed.
+        self._unification_max_seq_len = max_seq_len
         self._max_batch_size = max_batch_size
         self._net_max_seq_len = net_max_seq_len
         self._dummy_reqs = None
@@ -1148,6 +1153,11 @@ def _per_manager_cache_cost(self,
                            self._kv_cache_config)
         if tokens_per_block is None:
             tokens_per_block = self._tokens_per_block
+        if not is_draft and self._should_unify_dflash_kv_cache():
+            extra_kwargs = dict(extra_kwargs)
+            extra_kwargs[
+                "layer_specs"] = self._build_unified_dflash_layer_specs(
+                    kv_cache_config)
         if not is_draft:
             derived_windows = _derive_v2_layer_type_attention_windows(
                 kv_cache_config, manager_cls, model_config, self._max_seq_len)
@@ -2113,6 +2123,14 @@ def _create_kv_cache_manager(
             spec_dec_layer_mask = [True] * num_target_layers
 
         estimating_kv_cache = estimating_kv_cache and not self._skip_est
+        unified_layer_specs = None
+        if self._should_unify_dflash_kv_cache():
+            unified_layer_specs = self._build_unified_dflash_layer_specs(
+                kv_cache_config)
+            logger.info(
+                "DFlash: target and draft KV layers share one KV cache manager "
+                f"({len(unified_layer_specs)} layers, disagg={self._is_disagg})."
+            )
         kv_cache_manager = _create_kv_cache_manager(
             model_engine=self._model_engine,
             kv_cache_manager_cls=kv_cache_manager_cls,
@@ -2137,6 +2155,7 @@ def _create_kv_cache_manager(
             self._llm_args.kv_cache_config.kv_events_config,
             cold_page_codec_provider=cold_page_codec_provider,
             joint_kv_cache_reuse=self._joint_kv_cache_reuse,
+            layer_specs=unified_layer_specs,
         )
 
         if not self._skip_est:
@@ -2166,6 +2185,8 @@ def _should_create_separate_draft_kv_cache(self) -> bool:
         if self._speculative_config is None:
             # No drafter at all, so there is nothing to give a manager to.
             return False
+        if self._should_unify_dflash_kv_cache():
+            return False
         # Narrower than is_external_drafter(): PARD and DRAFT_TARGET_ONE_MODEL
         # never reach the arena this carve-out exists for.
         spec_dec_mode = self._speculative_config.spec_dec_mode
@@ -2191,6 +2212,198 @@ def _should_create_separate_draft_kv_cache(self) -> bool:
             return False
         return should_use_separate_draft_kv_cache(self._speculative_config)
 
+    def _should_unify_dflash_kv_cache(self) -> bool:
+        """Whether this run can place target and DFlash layers in one V2 manager."""
+        spec_config = self._speculative_config
+        if (spec_config is None or not spec_config.spec_dec_mode.is_dflash()
+                or spec_config.attention_backend != "TRTLLM"
+                or self._tokens_per_block not in (16, 32, 64)
+                or not spec_config._allow_separate_draft_kv_cache
+                or spec_config.draft_len_schedule is not None
+                or self._draft_config is None or self._mapping.pp_size != 1
+                or self._mapping.cp_size != 1
+                or self._llm_args.attn_backend != "TRTLLM"
+                or _has_variable_kv_cache_geometry(
+                    self._model_engine.model.model_config)
+                or getattr(self._kv_cache_config, "mixed_precision_kv",
+                           None) is not None
+                or self._llm_args.kv_cache_compression_config is not None):
+            return False
+        # Specialized managers add cache roles and lifecycle protocols that an
+        # external drafter cannot yet describe. Keep their existing fallback.
+        if self._kv_cache_manager_cls is not KVCacheManagerV2:
+            return False
+        target_layout = self._dflash_kv_layout(
+            self._model_engine.model.model_config,
+            self._model_engine.model.config.torch_dtype)
+        draft_model_config = self._get_draft_kv_model_config()
+        draft_layout = self._dflash_kv_layout(
+            draft_model_config,
+            draft_model_config.pretrained_config.torch_dtype)
+        if target_layout is None or draft_layout is None:
+            logger.info_once(
+                "DFlash: a per-layer KV-head count is not supported in a shared "
+                "target and draft KV cache; keeping a separate draft KV cache.",
+                key="dflash_per_layer_kv_heads_fallback")
+            return False
+        draft_windows = _resolve_v2_max_attention_window_vec(
+            self._get_draft_max_attention_window(self._unification_max_seq_len,
+                                                 self._kv_cache_config),
+            self._unification_max_seq_len,
+            range(draft_model_config.pretrained_config.num_hidden_layers))
+        if len(set(draft_windows)) > 1:
+            logger.info_once(
+                "DFlash: draft layers have different effective attention "
+                "windows; keeping a separate draft KV cache.",
+                key="dflash_mixed_draft_windows_fallback")
+            return False
+        if self._is_disagg and os.environ.get(
+                "TRTLLM_DISAGG_SHARED_DRAFT_KV") != "1":
+            return False
+        if self._is_disagg and not self._dflash_disagg_layouts_compatible(
+                target_layout, draft_layout):
+            logger.info_once(
+                "DFlash: draft KV layout (kv_heads, head_dim, dtype, k_only)="
+                f"{draft_layout} cannot be transferred with the target's "
+                f"{target_layout}; disaggregated serving keeps a separate "
+                "draft KV cache.",
+                key="dflash_disagg_draft_layout_fallback")
+            return False
+        return True
+
+    @staticmethod
+    def _dflash_disagg_layouts_compatible(
+            target_layout: tuple[int, int, object, bool],
+            draft_layout: tuple[int, int, object, bool]) -> bool:
+        """Whether the transceiver can move draft layers with the target's.
+
+        Heads are resharded in bytes: a view's bytes per head are its own
+        per-layer bytes over the rank's KV-head count, so dtype and head_dim
+        may differ. The KV-head count and K-only flag come from the rank's
+        single AttentionInfo and must match. NVFP4 adds block-scale views, so
+        an NVFP4 side is only paired with an identical layout.
+        """
+        target_heads, _, target_dtype, target_k_only = target_layout
+        draft_heads, _, draft_dtype, draft_k_only = draft_layout
+        if (target_heads, target_k_only) != (draft_heads, draft_k_only):
+            return False
+        nvfp4 = tensorrt_llm.bindings.DataType.NVFP4
+        return target_dtype == draft_dtype or nvfp4 not in (target_dtype,
+                                                            draft_dtype)
+
+    def _dflash_kv_layout(
+            self, model_config: ModelConfig,
+            fallback_dtype) -> Optional[tuple[int, int, object, bool]]:
+        """Rank-local (kv_heads, head_dim, dtype, k_only) of a model's KV layers.
+
+        Returns None when the KV-head count varies per layer.
+        """
+        config = model_config.pretrained_config
+        k_only = is_mla(config)
+        heads = 1 if k_only else getattr(config, "num_key_value_heads",
+                                         config.num_attention_heads)
+        if not isinstance(heads, int):
+            return None
+        head_dim = (config.kv_lora_rank + config.qk_rope_head_dim
+                    if k_only else getattr(config, "head_dim", None))
+        if not isinstance(head_dim, int):
+            head_dim = config.hidden_size // config.num_attention_heads
+        dtype = self._layer_spec_dtype(model_config, fallback_dtype)
+        # Attention DP replicates attention weights, so every rank holds all heads.
+        tp_size = 1 if self._mapping.enable_attention_dp else self._mapping.tp_size
+        return ceil_div(heads, tp_size), head_dim, dtype, k_only
+
+    @staticmethod
+    def _layer_spec_dtype(model_config: ModelConfig, fallback_dtype) -> object:
+        quant_config = model_config.quant_config
+        if quant_config is not None and quant_config.quant_mode.has_fp8_kv_cache(
+        ):
+            return tensorrt_llm.bindings.DataType.FP8
+        if quant_config is not None and quant_config.quant_mode.has_fp4_kv_cache(
+        ):
+            return tensorrt_llm.bindings.DataType.NVFP4
+        return str_dtype_to_binding(torch_dtype_to_str(fallback_dtype))
+
+    @staticmethod
+    def _layer_spec_roles(dtype, *, k_only: bool) -> tuple:
+        roles = [Role.KEY]
+        if not k_only:
+            roles.append(Role.VALUE)
+        if dtype == tensorrt_llm.bindings.DataType.NVFP4:
+            roles.append(Role.KEY_BLOCK_SCALE)
+            if not k_only:
+                roles.append(Role.VALUE_BLOCK_SCALE)
+        return tuple(roles)
+
+    def _build_unified_dflash_layer_specs(
+            self,
+            kv_cache_config: KvCacheConfig) -> tuple[KVCacheLayerSpec, ...]:
+        """Build the rank-local target+DFlash storage layout used by cost and allocation."""
+        if not self._should_unify_dflash_kv_cache():
+            return ()
+
+        target_model_config = self._model_engine.model.model_config
+        target_config = target_model_config.pretrained_config
+        target_layers = int(target_config.num_hidden_layers)
+        target_local_ids = list(self._mapping.pp_layers(target_layers))
+        target_heads, target_head_dim, target_dtype, target_k_only = (
+            self._dflash_kv_layout(target_model_config,
+                                   self._model_engine.model.config.torch_dtype))
+        target_window_config = kv_cache_config.max_attention_window
+        if target_window_config is None:
+            target_window_config = _derive_v2_layer_type_attention_windows(
+                kv_cache_config,
+                self._kv_cache_manager_cls,
+                target_model_config,
+                self._max_seq_len,
+            )
+        target_windows = _resolve_v2_max_attention_window_vec(
+            target_window_config,
+            self._max_seq_len,
+            target_local_ids,
+        )
+
+        draft_model_config = self._get_draft_kv_model_config()
+        draft_config = draft_model_config.pretrained_config
+        draft_layers = int(draft_config.num_hidden_layers)
+        draft_local_ids = list(self._mapping.pp_layers(draft_layers))
+        draft_heads, draft_head_dim, draft_dtype, draft_k_only = (
+            self._dflash_kv_layout(draft_model_config,
+                                   draft_config.torch_dtype))
+        draft_windows = _resolve_v2_max_attention_window_vec(
+            self._get_draft_max_attention_window(self._max_seq_len,
+                                                 kv_cache_config),
+            self._max_seq_len,
+            draft_local_ids,
+        )
+
+        specs = [
+            KVCacheLayerSpec(
+                global_layer_id=layer_id,
+                owner=KVCacheLayerOwner.TARGET,
+                dtype=target_dtype,
+                num_kv_heads=target_heads,
+                head_dim=target_head_dim,
+                buffer_roles=self._layer_spec_roles(target_dtype,
+                                                    k_only=target_k_only),
+                sliding_window_size=window,
+                reuse_policy=AttentionReusePolicy.REQUIRED,
+            ) for layer_id, window in zip(target_local_ids, target_windows)
+        ]
+        specs.extend(
+            KVCacheLayerSpec(
+                global_layer_id=target_layers + layer_id,
+                owner=KVCacheLayerOwner.DRAFT,
+                dtype=draft_dtype,
+                num_kv_heads=draft_heads,
+                head_dim=draft_head_dim,
+                buffer_roles=self._layer_spec_roles(draft_dtype,
+                                                    k_only=draft_k_only),
+                sliding_window_size=window,
+                reuse_policy=AttentionReusePolicy.REQUIRED,
+            ) for layer_id, window in zip(draft_local_ids, draft_windows))
+        return tuple(specs)
+
     def _joint_reuse_supported(self) -> bool:
         """Span half of the pairing decision; ``build_managers`` ANDs it with
         ``has_separate_one_model_draft``. ``False`` = leave the run unpaired.
@@ -2268,13 +2481,12 @@ def _get_draft_kv_model_config(self) -> ModelConfig:
         if not (spec_dec_mode.is_dflash() or spec_dec_mode.is_dspark()):
             return effective_draft_config
         quant_config = getattr(effective_draft_config, "quant_config", None)
-        if quant_config is None or not (
-                quant_config.quant_mode.has_fp8_kv_cache()
-                or quant_config.quant_mode.has_fp4_kv_cache()):
+        if quant_config is None or not quant_config.quant_mode.has_kv_cache_quant(
+        ):
             return effective_draft_config
         logger.info(
             "External drafter KV pool keeps the drafter dtype; dropping "
-            "the fp8/nvfp4 KV quant algo inherited from the target.")
+            "the KV quantization inherited from the target.")
         neutral_quant = copy.copy(quant_config)
         neutral_quant.kv_cache_quant_algo = None
         # QuantConfig.quant_mode and .layer_quant_mode are both cached_property
@@ -3327,7 +3539,9 @@ def _create_kv_cache_manager(
         cold_page_codec_provider: Optional[object] = None,
         kv_events_config: Optional[KVEventsConfig] = None,
         joint_kv_cache_reuse: bool = False,
-        max_cuda_graph_batch_size: Optional[int] = None) -> KVCacheManager:
+        max_cuda_graph_batch_size: Optional[int] = None,
+        layer_specs: Optional[Sequence[KVCacheLayerSpec]] = None
+) -> KVCacheManager:
     """
     Returns:
         A KVCacheManager instance for the given model engine or model config
@@ -3369,10 +3583,10 @@ def _create_kv_cache_manager(
 
     hidden_size = config.hidden_size
 
-    layer_specs = _get_kv_cache_layer_specs(_model_config)
-    if layer_specs:
-        head_dim = [spec.head_dim for spec in layer_specs]
-        num_key_value_heads = [spec.num_kv_heads for spec in layer_specs]
+    model_layer_specs = _get_kv_cache_layer_specs(_model_config)
+    if model_layer_specs:
+        head_dim = [spec.head_dim for spec in model_layer_specs]
+        num_key_value_heads = [spec.num_kv_heads for spec in model_layer_specs]
     else:
         num_attention_heads = config.num_attention_heads
         num_key_value_heads = num_kv_heads if num_kv_heads is not None else getattr(
@@ -3480,6 +3694,7 @@ def _create_kv_cache_manager(
         manager_extra_kwargs["joint_kv_cache_reuse"] = joint_kv_cache_reuse
         manager_extra_kwargs[
             "disable_overlap_scheduler"] = disable_overlap_scheduler
+        manager_extra_kwargs["layer_specs"] = layer_specs
         # Vocab size also enables multimodal event decoding and its per-block
         # digest scan. Leave it unset for text-only engines. Without an engine
         # (e.g. separate one-model draft caches), preserve config resolution:
diff --git a/tensorrt_llm/_torch/pyexecutor/scheduler/scheduler_v2.py b/tensorrt_llm/_torch/pyexecutor/scheduler/scheduler_v2.py
index 4cf174f17fd0..c9ce9f4b7fa6 100644
--- a/tensorrt_llm/_torch/pyexecutor/scheduler/scheduler_v2.py
+++ b/tensorrt_llm/_torch/pyexecutor/scheduler/scheduler_v2.py
@@ -1018,9 +1018,10 @@ def _try_schedule_disagg_gen_init(
         Returns ``(action, tokens)``.  *tokens* is 0 because disagg requests
         don't participate in the forward pass token budget.
         """
-        # Cache-transceiver mode disables the separate one-model draft manager,
-        # so disagg generation init has no paired-reuse path. Supporting one
-        # would also require draft KV transfer and history_length=prompt_len.
+        # Only the target manager receives prompt KV. Drafters that need it
+        # under disagg keep their layers in that manager (MTP/Eagle fold them
+        # in; DFlash uses explicit draft layer specs). A separate draft manager
+        # gets no transfer and no paired-reuse path here.
         if not self.kv_cache_manager.prepare_disagg_gen_init(req):
             logger.debug("prepare_disagg_gen_init failed for request %s", req.py_request_id)
             return ScheduleAction.SKIP, 0
diff --git a/tensorrt_llm/_torch/speculative/dflash.py b/tensorrt_llm/_torch/speculative/dflash.py
index fb85eafd3cd8..2a3ee089f04c 100644
--- a/tensorrt_llm/_torch/speculative/dflash.py
+++ b/tensorrt_llm/_torch/speculative/dflash.py
@@ -31,6 +31,7 @@
 
 from ..attention.backends import AttentionMetadata
 from ..modules.logits_processor import LogitsProcessor
+from ..pyexecutor.kv_cache.kv_cache_manager_v2 import KVCacheManagerV2, Role
 from ..pyexecutor.kv_cache.mamba_cache_manager import MambaHybridCacheManager
 from ..pyexecutor.llm_request import ATTENTION_DP_DUMMY_REQUEST_ID
 from ..pyexecutor.resource_manager import BaseResourceManager
@@ -615,19 +616,30 @@ def prepare(self):
             # prefilling, so _store_prefill_context -- the only other assigner --
             # never runs and every request would land on the dummy slot.
             num_contexts = max(0, len(self.request_ids) - self.num_generations)
-            for rid in self.request_ids[num_contexts:]:
+            assigned = {}
+            for request_idx, rid in enumerate(self.request_ids[num_contexts:], num_contexts):
                 if (
                     rid != ATTENTION_DP_DUMMY_REQUEST_ID
                     and rid < worker._graph_dummy_id_floor
                     and rid not in worker._req_to_slot
                 ):
-                    if worker._assign_slot(rid) is None:
+                    slot = worker._assign_slot(rid, updates=assigned)
+                    if slot is None:
                         # Not fatal -- spec dec is lossless, so the dummy-slot
                         # fallback only costs acceptance. Warn so it is not silent.
                         logger.warning(
                             f"DFlash: no free drafter slot for generation request {rid}; "
                             "it shares the dummy slot and will draft poorly."
                         )
+                    elif worker._ctx_unified_cache:
+                        # New requests have no in-flight predecessor. Subsequent
+                        # accepted tokens advance the device length during replay.
+                        token_offset = sum(self.seq_lens[:request_idx])
+                        assigned[slot] = min(
+                            int(self.host_position_ids[token_offset]), worker._max_ctx
+                        )
+            if assigned:
+                worker._write_ctx_len(assigned)
 
             # Route unknown request IDs (cuda graph padding or warmup dummies)
             # to dummy slot to avoid corrupting real request's context
@@ -794,6 +806,9 @@ def __init__(
         # the table width on the default backend -- dflash_allocated_ctx_limit.
         self._ctx_block_counts = None
         self._ctx_pool_idx = 0
+        self._ctx_layer_ids = None
+        self._ctx_unified_cache = False
+        self._ctx_block_offset = 0
         self._ctx_block_divisor = 1  # encoded offset -> raw pool block index
         self._ctx_kv_indptr = None
         self._ctx_kv_last_page_len = None
@@ -1083,7 +1098,18 @@ def _managed_ctx_pool(self, draft_kv_cache_manager, L, nkv, hd, dtype, kv_factor
             # it: _util.py skips that bail for an external drafter, which is
             # what DSpark is, so this drafter does get a manager under DP.
             return None
-        layers = [draft_kv_cache_manager.get_buffers(i, kv_layout="HND") for i in range(L)]
+        draft_layer_ids = tuple(getattr(draft_kv_cache_manager, "draft_layer_ids", ()))
+        layer_ids = draft_layer_ids if draft_layer_ids else tuple(range(L))
+        if len(layer_ids) != L:
+            logger.warning(
+                f"DFlash: KV cache manager exposes {len(layer_ids)} draft layers "
+                f"but the drafter has {L}; using the private ctx arena."
+            )
+            return None
+        self._ctx_layer_ids = layer_ids
+        layers = [
+            draft_kv_cache_manager.get_buffers(layer_id, kv_layout="HND") for layer_id in layer_ids
+        ]
         base = layers[0]
         # kv_factor 1 is the MLA drafter's SELFKONLY pool: one latent, no V.
         expected = (kv_factor, nkv, base.size(-2), hd)
@@ -1116,10 +1142,40 @@ def _init_ctx_block_tables(
         """
         mgr = draft_kv_cache_manager
         pool_mapping = mgr.kv_cache_pool_mapping
-        layer_offset = mgr.layer_offsets[0]
+        first_layer_id = self._ctx_layer_ids[0] if self._ctx_layer_ids is not None else 0
+        layer_offset = mgr.layer_offsets[first_layer_id]
         self._ctx_pool_idx = int(pool_mapping[layer_offset, 0])
+        per_layer_tables = bool(getattr(mgr, "_use_per_layer_page_tables", False))
         num_pool_layers = int((pool_mapping[:, 0] == self._ctx_pool_idx).sum())
-        if num_pool_layers != L:
+        self._ctx_block_offset = 0
+        if isinstance(mgr, KVCacheManagerV2):
+            converter = mgr.impl.get_page_index_converter(layer_offset, Role.KEY)
+            layer_group_id = mgr.layer_to_pool_mapping_dict[layer_offset]
+            if mgr.enable_swa_scratch_reuse:
+                raise NotImplementedError(
+                    "DFlash managed context does not support SWA scratch pages"
+                )
+            for layer_id in self._ctx_layer_ids:
+                local_id = mgr.layer_offsets[layer_id]
+                if (
+                    mgr.layer_to_pool_mapping_dict[local_id] != layer_group_id
+                    or mgr.get_layer_page_index_scale(layer_id) != converter.scale
+                ):
+                    if self._ctx_unified_cache:
+                        raise NotImplementedError(
+                            "DFlash's shared context block table requires draft layers "
+                            "with the same lifecycle and page-index scale"
+                        )
+                    logger.warning(
+                        "DFlash: draft KV layers have different lifecycles or page-index "
+                        "scales; using the private ctx arena."
+                    )
+                    return False
+            if per_layer_tables:
+                # Attention tables address the physical pool origin, whereas
+                # get_buffers() starts at this layer's own K buffer.
+                self._ctx_block_offset = int(converter.layer_offset)
+        elif num_pool_layers != L:
             logger.warning(
                 f"DFlash: draft KV pool {self._ctx_pool_idx} spans {num_pool_layers} layers but "
                 f"the drafter has {L}; using the private ctx arena."
@@ -1128,12 +1184,23 @@ def _init_ctx_block_tables(
         # V1 slices a page-major pool, so stride(0) spans every layer; V2 wraps
         # the layer base in a dense view over interleaved pages, so stride(0) is
         # one layer slot. Guessing wrong is silent: acceptance quietly drops.
-        layer_block_elems = mgr.kv_factor * nkv * page_size * hd
+        manager_kv_factor = (
+            mgr.get_layer_kv_factor(first_layer_id)
+            if hasattr(mgr, "get_layer_kv_factor")
+            else mgr.kv_factor
+        )
+        if manager_kv_factor != getattr(self, "_ctx_kv_factor", manager_kv_factor):
+            logger.warning(
+                "DFlash: draft KV layer factor does not match the drafter; "
+                "using the private ctx arena."
+            )
+            return False
+        layer_block_elems = manager_kv_factor * nkv * page_size * hd
         stride0 = pool[0].stride(0)
         if stride0 == layer_block_elems:
-            self._ctx_block_divisor = mgr.kv_factor
+            self._ctx_block_divisor = manager_kv_factor
         elif stride0 == num_pool_layers * layer_block_elems:
-            self._ctx_block_divisor = num_pool_layers * mgr.kv_factor
+            self._ctx_block_divisor = num_pool_layers * manager_kv_factor
         else:
             logger.warning(
                 f"DFlash: draft KV pool layer stride {stride0} matches neither V1's "
@@ -1166,14 +1233,17 @@ def _refresh_ctx_block_tables(self, attn_metadata, num_seqs: int) -> bool:
         """
         if self._ctx_block_tables is None or num_seqs <= 0:
             return False
-        src = getattr(attn_metadata, "draft_kv_cache_block_offsets", None)
+        offsets_name = (
+            "kv_cache_block_offsets" if self._ctx_unified_cache else "draft_kv_cache_block_offsets"
+        )
+        src = getattr(attn_metadata, offsets_name, None)
         if src is None:
             # The table is bound but unfillable. Leaving it at zeros would send
             # every request's every layer to pool block 0 -- silently.
             raise RuntimeError(
                 "DFlash: bound to the draft KV pool but "
                 f"{type(attn_metadata).__name__} carries no "
-                "draft_kv_cache_block_offsets this iteration."
+                f"{offsets_name} this iteration."
             )
         # Field 0 of the K/V pair: both name the same block for a kv_factor=2
         # pool, and the drafter addresses blocks, not planes. The block count
@@ -1185,6 +1255,7 @@ def _refresh_ctx_block_tables(self, attn_metadata, num_seqs: int) -> bool:
             self._ctx_block_divisor,
             self._ctx_block_tables,
             self._ctx_block_counts,
+            block_offset=self._ctx_block_offset,
         )
         return True
 
@@ -1192,12 +1263,18 @@ def _lazy_init_ctx_buffers(
         self, draft_model, spec_metadata, attn_metadata, draft_kv_cache_manager=None
     ):
         self._validate_draft_attention_backend(draft_model)
-        # Only TrtllmAttentionMetadata builds draft_kv_cache_block_offsets.
-        # Without it the table stays all-zero and every request would read and
-        # write pool block 0 -- silently, at acceptance 1.0.
-        if draft_kv_cache_manager is not None and not hasattr(
-            attn_metadata, "draft_kv_cache_block_offsets"
-        ):
+        self._ctx_unified_cache = (
+            draft_kv_cache_manager is not None
+            and draft_kv_cache_manager is getattr(attn_metadata, "kv_cache_manager", None)
+        )
+        offsets_name = (
+            "kv_cache_block_offsets" if self._ctx_unified_cache else "draft_kv_cache_block_offsets"
+        )
+        if draft_kv_cache_manager is not None and not hasattr(attn_metadata, offsets_name):
+            if self._ctx_unified_cache:
+                raise NotImplementedError(
+                    f"DFlash unified KV requires {offsets_name} in attention metadata"
+                )
             logger.warning(
                 f"DFlash: {type(attn_metadata).__name__} carries no draft KV block offsets; "
                 "using the private ctx arena."
@@ -1306,6 +1383,7 @@ def _lazy_init_ctx_buffers(
         nkv = draft_model._num_kv_heads
         hd = draft_model._head_dim
         kv_factor = getattr(draft_model, "_kv_factor", 2)
+        self._ctx_kv_factor = kv_factor
         capacity = self._max_ctx + self._compute_block_size
         # Published before any drafter table is built, so a drafter may read it
         # instead of its config cap. _max_ctx is attn_metadata.max_seq_len, which
@@ -1361,6 +1439,8 @@ def _lazy_init_ctx_buffers(
             ):
                 pool = None
             if pool is None:
+                if self._ctx_unified_cache:
+                    raise ValueError("DFlash could not bind the unified manager's draft KV layers")
                 # Per-slot reservation is an arena-only notion; the pool path
                 # takes its pages from the manager one iteration at a time.
                 self._ctx_pages_per_slot = (capacity + page_size - 1) // page_size
@@ -1399,7 +1479,7 @@ def _lazy_init_ctx_buffers(
                 # Only a manager that publishes and matches its own blocks can
                 # hand back a reused prefix's drafter K/V; the private arena and
                 # an unpaired draft pool both start every request at 0.
-                self._ctx_reuse_addressable = bool(
+                self._ctx_reuse_addressable = self._ctx_unified_cache or bool(
                     getattr(draft_kv_cache_manager, "enable_joint_kv_cache_reuse", False)
                 )
             self._ctx_kv_last_page_len = torch.full(
@@ -1819,10 +1899,18 @@ def _forward_impl(
 
         # Lazy init buffers and attach worker reference for prepare()
         draft_kv_cache_manager = self.get_draft_kv_cache_manager(resource_manager)
+        previous_ctx_len = self._ctx_len
         self._lazy_init_ctx_buffers(
             draft_model, spec_metadata, attn_metadata, draft_kv_cache_manager
         )
         spec_metadata._dflash_worker = self
+        if (
+            self._ctx_unified_cache
+            and self._ctx_len is not previous_ctx_len
+            and spec_metadata.request_ids
+        ):
+            # New buffers clear request slots; seed them before generation.
+            spec_metadata.prepare()
         # Before any store: prefill and decode both address pages through it.
         # Returning False here means an empty batch -- the missing-offsets case
         # raises inside, with the metadata type in the message.
@@ -2342,7 +2430,7 @@ def prepare_1st_drafter_inputs(
                     projected_rows = projected_to_store.reshape(num_gens, K_plus_1, -1)
                     lilicorr_anchor = last_accepted_hidden(projected_rows, gen_num_accepted)
                 gen_num_accepted_long = gen_num_accepted.long()
-                col_idx = self._ctx_len[slots].unsqueeze(1) + offsets_kp1.unsqueeze(0)
+                col_idx = ctx_len_gen.unsqueeze(1) + offsets_kp1.unsqueeze(0)
                 write_mask = offsets_kp1.unsqueeze(0) < gen_num_accepted_long.unsqueeze(1)
                 if self._ctx_block_tables is not None:
                     # Bound by what the manager handed THIS request: the masked
diff --git a/tensorrt_llm/_torch/speculative/interface.py b/tensorrt_llm/_torch/speculative/interface.py
index 97b67a555523..f5c6c1aa82d9 100644
--- a/tensorrt_llm/_torch/speculative/interface.py
+++ b/tensorrt_llm/_torch/speculative/interface.py
@@ -179,7 +179,8 @@ def prepare_attn_metadata_for_draft_replay(attn_metadata,
     Call restore_attn_metadata_after_draft_replay in a finally block.
     Returns saved state or None if no-op.
     """
-    if draft_kv_cache_manager is None:
+    if (draft_kv_cache_manager is None
+            or draft_kv_cache_manager is attn_metadata.kv_cache_manager):
         return None
     if not isinstance(attn_metadata, TrtllmAttentionMetadata):
         return None
@@ -3239,8 +3240,10 @@ def get_draft_kv_cache_manager(self, resource_manager):
         Get the draft KV cache manager if using separate KV cache layouts.
         """
         if self.use_separate_draft_kv_cache and resource_manager is not None:
-            return resource_manager.get_resource_manager(
+            draft = resource_manager.get_resource_manager(
                 ResourceManagerType.DRAFT_KV_CACHE_MANAGER)
+            if draft is not None:
+                return draft
         if resource_manager is not None:
             target = resource_manager.get_resource_manager(
                 ResourceManagerType.KV_CACHE_MANAGER)
@@ -3261,7 +3264,8 @@ def draft_kv_cache_context(self, attn_metadata, draft_kv_cache_manager):
         """
 
         # draft_kv_cache_manager is None if using two-engine speculative decoding or not enabling separate draft KV cache.
-        if draft_kv_cache_manager is None:
+        if (draft_kv_cache_manager is None
+                or draft_kv_cache_manager is attn_metadata.kv_cache_manager):
             yield attn_metadata
             return
 
diff --git a/tensorrt_llm/_torch/speculative/spec_tail_kernels.py b/tensorrt_llm/_torch/speculative/spec_tail_kernels.py
index 6705f605a9eb..746d9631bfff 100644
--- a/tensorrt_llm/_torch/speculative/spec_tail_kernels.py
+++ b/tensorrt_llm/_torch/speculative/spec_tail_kernels.py
@@ -134,6 +134,7 @@ def _decode_block_table_kernel(
     stride_src_col,
     n_blocks,
     divisor,
+    block_offset,
     table_ptr,
     stride_table,
     counts_ptr,
@@ -148,21 +149,32 @@ def _decode_block_table_kernel(
         cols = start + offs
         mask = cols < n_blocks
         encoded = tl.load(src_row + cols * stride_src_col, mask=mask, other=-1).to(tl.int64)
+        encoded -= block_offset
         count += (encoded >= 0).to(tl.int64)
         tl.store(table_row + cols, (tl.maximum(encoded, 0) // divisor).to(tl.int32), mask=mask)
     tl.store(counts_ptr + row, tl.sum(count, axis=0))
 
 
 def decode_draft_block_table(
-    src: torch.Tensor, divisor: int, table: torch.Tensor, counts: torch.Tensor
+    src: torch.Tensor,
+    divisor: int,
+    table: torch.Tensor,
+    counts: torch.Tensor,
+    *,
+    block_offset: int = 0,
 ) -> None:
-    """``counts[i] = (src[i] >= 0).sum()``; ``table[i] = src[i].clamp(min=0) // divisor``.
+    """Decode layer-relative block offsets without modifying ``src``.
+
+    For ``encoded = src - block_offset``, computes
+    ``counts[i] = (encoded[i] >= 0).sum()`` and
+    ``table[i] = encoded[i].clamp(min=0) // divisor``.
 
     Args:
         src: ``[rows, blocks]`` encoded block offsets (any strides).
         divisor: encoded offset -> pool block index divisor.
         table: ``[>= rows, blocks]`` int32 output, unit inner stride.
         counts: ``[>= rows]`` int64 output.
+        block_offset: this layer's K-buffer offset within each physical slot.
     """
     rows, n_blocks = src.shape
     assert table.shape[-1] == n_blocks and table.stride(-1) == 1
@@ -174,6 +186,7 @@ def decode_draft_block_table(
         src.stride(1),
         n_blocks,
         int(divisor),
+        int(block_offset),
         table,
         table.stride(0),
         counts,
diff --git a/tensorrt_llm/_torch/speculative/utils.py b/tensorrt_llm/_torch/speculative/utils.py
index f8e1a7340145..b779f24e2a0a 100644
--- a/tensorrt_llm/_torch/speculative/utils.py
+++ b/tensorrt_llm/_torch/speculative/utils.py
@@ -804,7 +804,9 @@ def get_draft_kv_cache_manager(spec_config, resource_manager):
     target = resource_manager.get_resource_manager(
         ResourceManagerType.KV_CACHE_MANAGER)
     get_view = getattr(target, "get_draft_kv_cache_view", None)
-    return get_view() if get_view is not None else None
+    view = get_view() if get_view is not None else None
+    # A unified manager's request resources are already handled through target.
+    return None if view is target else view
 
 
 def update_spec_config_from_model_config(spec_config,
diff --git a/tensorrt_llm/commands/serve.py b/tensorrt_llm/commands/serve.py
index 1a0f80f8c3b5..1c3243ac70b1 100644
--- a/tensorrt_llm/commands/serve.py
+++ b/tensorrt_llm/commands/serve.py
@@ -2721,6 +2721,37 @@ def disaggregated_mpi_worker(config_file: Optional[str], log_level: str,
     from tensorrt_llm.llmapi.disagg_utils import split_world_comm
 
     disagg_cfg = parse_disagg_config_file(config_file)
+    with_spec = {
+        cfg.type
+        for cfg in disagg_cfg.server_configs
+        if cfg.other_args.get("speculative_config")
+    }
+    attention_parallel_configs = {(cfg.other_args.get("tensor_parallel_size",
+                                                      1),
+                                   cfg.other_args.get("enable_attention_dp",
+                                                      False))
+                                  for cfg in disagg_cfg.server_configs}
+    shared_draft_kv_supported = True
+    for cfg in disagg_cfg.server_configs:
+        spec = cfg.other_args.get("speculative_config") or {}
+        kv_cache = cfg.other_args.get("kv_cache_config") or {}
+        if (spec.get("decoding_type") != "DFlash"
+                or spec.get("attention_backend") != "TRTLLM"
+                or spec.get("draft_len_schedule") is not None
+                or spec.get("max_concurrency") is not None
+                or cfg.other_args.get("attn_backend", "TRTLLM") != "TRTLLM"
+                or cfg.other_args.get("pipeline_parallel_size", 1) != 1
+                or cfg.other_args.get("context_parallel_size", 1) != 1
+                or cfg.other_args.get("kv_cache_compression_config") is not None
+                or str(kv_cache.get("dtype",
+                                    "auto")).lower().startswith("mixed_")
+                or kv_cache.get("tokens_per_block", 32) not in (16, 32, 64)):
+            shared_draft_kv_supported = False
+            break
+    os.environ["TRTLLM_DISAGG_SHARED_DRAFT_KV"] = (
+        "1"
+        if with_spec == {"ctx", "gen"} and len(attention_parallel_configs) == 1
+        and shared_draft_kv_supported else "0")
 
     # Run a server with the underlying LLM invokes a RemoteMPISessionClient
     if os.environ.get(DisaggLauncherEnvs.
diff --git a/tests/integration/defs/accuracy/references/acceptance_length.yaml b/tests/integration/defs/accuracy/references/acceptance_length.yaml
index 97cedd3d9a2d..fc3eaad3ac9b 100644
--- a/tests/integration/defs/accuracy/references/acceptance_length.yaml
+++ b/tests/integration/defs/accuracy/references/acceptance_length.yaml
@@ -137,6 +137,12 @@ ModelingV2DeepseekR10528Nvfp4Sm10xDefault::mtp3:
 TestGLM53FlashFP8::test_mtp:
   ref_al: 3.384
   min_al: 3.215
+# Measured on B200, 1P1D TP1 with target and draft KV in one manager.
+# min_al is 95% of the measured value.
+TestQwen3_8BDFlash::test_gsm8k_1p1d_unified_draft_kv:
+  ref_al: 4.062
+  min_al: 3.859
+
 # Read by TestModelingV2Glm52Nvfp4Sm103Tep4::
 # test_gsm8k_identity_vs_mtp1_and_acceptance. Keyed by target and variant for
 # the same reason as the ModelingV2DeepseekR10528Nvfp4Sm10xDefault::mtp3 entry
diff --git a/tests/integration/defs/accuracy/test_disaggregated_serving.py b/tests/integration/defs/accuracy/test_disaggregated_serving.py
index 4bf85c4041ce..030292c22ae0 100644
--- a/tests/integration/defs/accuracy/test_disaggregated_serving.py
+++ b/tests/integration/defs/accuracy/test_disaggregated_serving.py
@@ -39,7 +39,8 @@
 
 from tensorrt_llm.executor.result import GenerationResultBase
 from tensorrt_llm.llmapi import CompletionOutput, RequestOutput, SamplingParams
-from tensorrt_llm.llmapi.llm_args import (DSparkDecodingConfig,
+from tensorrt_llm.llmapi.llm_args import (DFlashDecodingConfig,
+                                          DSparkDecodingConfig,
                                           Eagle3DecodingConfig, LlmArgs,
                                           MTPDecodingConfig)
 from tensorrt_llm.llmapi.tokenizer import load_hf_tokenizer
@@ -1648,6 +1649,96 @@ def test_gen_first(self, enable_block_reuse):
             run_accuracy_test(llm, self.MODEL_NAME, ["MMLU"])
 
 
+@skip_pre_blackwell
+@pytest.mark.timeout(DEFAULT_TEST_TIMEOUT)
+class TestQwen3_8BDFlash(LlmapiAccuracyTestHarness):
+    MODEL_NAME = "Qwen3/Qwen3-8B"
+    MODEL_PATH = f"{llm_models_root()}/Qwen3/Qwen3-8B"
+    DRAFT_PATH = f"{llm_models_root()}/Qwen3-8B-DFlash-b16"
+    MAX_DRAFT_LEN = 15
+
+    @pytest.mark.skip_less_device(2)
+    def test_gsm8k_1p1d_unified_draft_kv(self):
+        """DFlash draft KV travels with the target KV from ctx to gen.
+
+        The drafter's layers share the target's KV cache manager, so the
+        generation worker drafts from the prompt context computed on the
+        context worker. Verified speculative decoding is lossless, so the
+        GSM8K score cannot see a drafter that starts from an empty context;
+        acceptance length can, so it is gated too.
+        """
+        speculative_config = {
+            "decoding_type": "DFlash",
+            "max_draft_len": self.MAX_DRAFT_LEN,
+            "speculative_model": self.DRAFT_PATH,
+            # The unified draft KV layout requires the TRTLLM drafter backend.
+            "attention_backend": "TRTLLM",
+        }
+        # Both workers run the drafter: the context worker writes the
+        # drafter's prompt KV into the shared manager, which is then
+        # transferred with the target KV.
+        common_server_config = {
+            "disable_overlap_scheduler": True,
+            "max_batch_size": 16,
+            "max_seq_len": 4096,
+            "max_num_tokens": 4096,
+            "kv_cache_config": {
+                "enable_block_reuse": False,
+                "free_gpu_memory_fraction": 0.5,
+            },
+            # KVCacheManagerV2 requires the Python transceiver.
+            "cache_transceiver_config": {
+                "backend": "NIXL",
+                "transceiver_runtime": "PYTHON",
+            },
+            "speculative_config": speculative_config,
+        }
+        ctx_server_config = {**common_server_config, "cuda_graph_config": None}
+        gen_server_config = {
+            **common_server_config,
+            "cuda_graph_config": {
+                "max_batch_size": 16,
+                "enable_padding": True,
+            },
+            # Acceptance length is read back off this worker's /metrics.
+            "enable_iter_perf_stats": True,
+            "iter_stats_max_iterations": -1,
+            "max_stats_len": -1,
+        }
+        disaggregated_server_config = {
+            "hostname": "localhost",
+            "backend": "pytorch",
+            "context_servers": {
+                "num_instances": 1
+            },
+            "generation_servers": {
+                "num_instances": 1
+            },
+        }
+        with launch_disaggregated_llm(
+                disaggregated_server_config,
+                ctx_server_config,
+                gen_server_config,
+                self.MODEL_PATH,
+                extra_env={"TRTLLM_DISAGG_SHARED_DRAFT_KV": "1"},
+                assert_worker_log_contains=
+                "target and draft KV layers share one KV cache manager") as llm:
+            llm.args.speculative_config = DFlashDecodingConfig(
+                max_draft_len=self.MAX_DRAFT_LEN,
+                speculative_model=self.DRAFT_PATH,
+            )
+            run_accuracy_test(llm, self.MODEL_NAME, ["GSM8K"])
+
+            # Drafting from an empty prompt context costs ~17% acceptance
+            # length on this model, well below the 95% floor.
+            acceptance_length = compute_disagg_acceptance_length(llm.serve_url)
+            print(f"[AL] test_gsm8k_1p1d_unified_draft_kv "
+                  f"acceptance_length = {acceptance_length:.3f}")
+            assert_acceptance_length(
+                "TestQwen3_8BDFlash::test_gsm8k_1p1d_unified_draft_kv",
+                acceptance_length)
+
+
 @skip_pre_blackwell
 @pytest.mark.timeout(DEFAULT_TEST_TIMEOUT)
 class TestQwen3_30B_A3B(LlmapiAccuracyTestHarness):
diff --git a/tests/integration/test_lists/test-db/l0_dgx_b200.yml b/tests/integration/test_lists/test-db/l0_dgx_b200.yml
index 5dfdc6fbcdd4..17cff489f10a 100644
--- a/tests/integration/test_lists/test-db/l0_dgx_b200.yml
+++ b/tests/integration/test_lists/test-db/l0_dgx_b200.yml
@@ -31,6 +31,7 @@ l0_dgx_b200:
   - accuracy/test_llm_api_pytorch.py::TestQwen3_5_35B_A3B::test_bf16[tp2-TRTLLM]
   - disaggregated/test_disaggregated_beam_search.py::test_disaggregated_beam_search_v2[beam2-reuse]
   - disaggregated/test_disaggregated_beam_search.py::test_disaggregated_beam_search_v2[beam4-no-reuse]
+  - accuracy/test_disaggregated_serving.py::TestQwen3_8BDFlash::test_gsm8k_1p1d_unified_draft_kv
   # ------------- KV Cache V2 Scheduler IT (multi-GPU) ---------------
   - kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2DSv3Lite::test_mtp_draft_tokens
   - kv_cache/test_kv_cache_v2_scheduler.py::TestKVCacheV2DSv3Lite::test_mtp_chunked_draft_tokens
diff --git a/tests/unittest/_torch/pyexecutor/kv_cache/test_dflash_unified_kv_gate.py b/tests/unittest/_torch/pyexecutor/kv_cache/test_dflash_unified_kv_gate.py
new file mode 100644
index 000000000000..8e128f058da8
--- /dev/null
+++ b/tests/unittest/_torch/pyexecutor/kv_cache/test_dflash_unified_kv_gate.py
@@ -0,0 +1,415 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Selection of the unified target+DFlash KV cache layout."""
+
+import os
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import torch
+import yaml
+
+from tensorrt_llm._torch.model_config import KVCacheLayerSpec
+from tensorrt_llm._torch.pyexecutor._util import KvCacheCreator
+from tensorrt_llm._torch.pyexecutor.kv_cache.kv_cache_manager_v2 import KVCacheManagerV2
+from tensorrt_llm.bindings import DataType
+from tensorrt_llm.quantization import QuantAlgo
+from tensorrt_llm.quantization.mode import QuantMode
+
+pytestmark = pytest.mark.cpu_only
+
+_SHARED_DRAFT_KV_ENV = "TRTLLM_DISAGG_SHARED_DRAFT_KV"
+
+
+@pytest.fixture(autouse=True)
+def _shared_draft_kv_allowed(monkeypatch):
+    # Tests of the layout rules assume a launcher that allows sharing.
+    monkeypatch.setenv(_SHARED_DRAFT_KV_ENV, "1")
+
+
+def _model_config(*, kv_heads, head_dim, kv_quant_algo=None):
+    pretrained = SimpleNamespace(
+        num_hidden_layers=2,
+        num_attention_heads=32,
+        num_key_value_heads=kv_heads,
+        head_dim=head_dim,
+        hidden_size=32 * head_dim,
+        torch_dtype=torch.bfloat16,
+    )
+    quant_mode = QuantMode.from_quant_algo(kv_cache_quant_algo=kv_quant_algo)
+    return SimpleNamespace(
+        pretrained_config=pretrained,
+        quant_config=SimpleNamespace(quant_mode=quant_mode),
+    )
+
+
+def _creator(*, is_disagg, target, draft, tp_size=1):
+    creator = object.__new__(KvCacheCreator)
+    creator._speculative_config = SimpleNamespace(
+        spec_dec_mode=SimpleNamespace(is_dflash=lambda: True, is_external_drafter=lambda: True),
+        attention_backend="TRTLLM",
+        _allow_separate_draft_kv_cache=True,
+        draft_len_schedule=None,
+    )
+    creator._draft_config = draft
+    creator._mapping = SimpleNamespace(
+        pp_size=1,
+        cp_size=1,
+        tp_size=tp_size,
+        enable_attention_dp=False,
+        pp_layers=lambda num_layers: range(num_layers),
+    )
+    creator._llm_args = SimpleNamespace(attn_backend="TRTLLM", kv_cache_compression_config=None)
+    creator._model_engine = SimpleNamespace(
+        model=SimpleNamespace(model_config=target, config=target.pretrained_config)
+    )
+    creator._kv_cache_manager_cls = KVCacheManagerV2
+    creator._kv_cache_config = SimpleNamespace(max_attention_window=None)
+    creator._tokens_per_block = 32
+    creator._unification_max_seq_len = 4096
+    creator._max_seq_len = 4096
+    creator._is_disagg = is_disagg
+    creator._get_draft_kv_model_config = lambda: draft
+    return creator
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+def test_matching_draft_layout_unifies(is_disagg):
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=8, head_dim=128)
+
+    creator = _creator(is_disagg=is_disagg, target=target, draft=draft)
+
+    assert creator._should_unify_dflash_kv_cache()
+    specs = creator._build_unified_dflash_layer_specs(creator._kv_cache_config)
+    assert [spec.global_layer_id for spec in specs] == [0, 1, 2, 3]
+    assert [spec.dtype for spec in specs] == [DataType.BF16] * 4
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+@pytest.mark.parametrize(
+    ("tokens_per_block", "draft_tokens_per_block"),
+    [(8, 32), (16, 32), (32, 32), (64, 64), (128, 32)],
+)
+def test_unification_requires_supported_draft_page_size(
+    is_disagg: bool, tokens_per_block: int, draft_tokens_per_block: int
+) -> None:
+    creator = _creator(
+        is_disagg=is_disagg,
+        target=_model_config(kv_heads=8, head_dim=128),
+        draft=_model_config(kv_heads=8, head_dim=128),
+    )
+    creator._tokens_per_block = tokens_per_block
+
+    assert creator._should_unify_dflash_kv_cache() is (tokens_per_block in (16, 32, 64))
+    assert creator._get_draft_tokens_per_block() == draft_tokens_per_block
+
+
+def test_mismatched_draft_kv_heads_unify_only_in_aggregated_serving() -> None:
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=4, head_dim=128)
+
+    assert _creator(is_disagg=False, target=target, draft=draft)._should_unify_dflash_kv_cache()
+    assert not _creator(is_disagg=True, target=target, draft=draft)._should_unify_dflash_kv_cache()
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+def test_mismatched_draft_head_dim_unifies(is_disagg: bool) -> None:
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=8, head_dim=64)
+
+    assert _creator(is_disagg=is_disagg, target=target, draft=draft)._should_unify_dflash_kv_cache()
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+def test_fp8_target_with_bf16_draft_unifies(is_disagg: bool) -> None:
+    target = _model_config(kv_heads=8, head_dim=128, kv_quant_algo=QuantAlgo.FP8)
+    draft = _model_config(kv_heads=8, head_dim=128)
+
+    creator = _creator(is_disagg=is_disagg, target=target, draft=draft)
+
+    assert creator._should_unify_dflash_kv_cache()
+    specs = creator._build_unified_dflash_layer_specs(creator._kv_cache_config)
+    assert [spec.dtype for spec in specs] == [DataType.FP8] * 2 + [DataType.BF16] * 2
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+def test_variable_model_geometry_keeps_separate_cache(is_disagg: bool) -> None:
+    target = _model_config(kv_heads=8, head_dim=128)
+    target.kv_cache_layer_specs = (
+        KVCacheLayerSpec(head_dim=128, num_kv_heads=8, num_q_heads=32),
+        KVCacheLayerSpec(head_dim=256, num_kv_heads=4, num_q_heads=32),
+    )
+    draft = _model_config(kv_heads=8, head_dim=128)
+
+    assert not _creator(
+        is_disagg=is_disagg, target=target, draft=draft
+    )._should_unify_dflash_kv_cache()
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+def test_retained_token_mixed_precision_keeps_separate_cache(is_disagg: bool) -> None:
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=8, head_dim=128)
+    creator = _creator(is_disagg=is_disagg, target=target, draft=draft)
+    creator._kv_cache_config.mixed_precision_kv = object()
+
+    assert not creator._should_unify_dflash_kv_cache()
+
+
+def test_layout_compares_rank_local_heads():
+    # 8 target heads and 2 draft heads differ per rank at every TP size; at
+    # TP4 the draft is duplicated to 1 head per rank while the target has 2.
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=2, head_dim=128)
+    creator = _creator(is_disagg=True, target=target, draft=draft, tp_size=4)
+
+    assert creator._dflash_kv_layout(target, torch.bfloat16)[0] == 2
+    assert creator._dflash_kv_layout(draft, torch.bfloat16)[0] == 1
+    assert not creator._should_unify_dflash_kv_cache()
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+def test_attention_dp_unifies_with_unsharded_heads(is_disagg):
+    # Under attention DP each rank keeps every KV head, so the TP4 layouts of
+    # an 8-head target and a 2-head draft are not resharded to 2 and 1.
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=8, head_dim=128)
+    creator = _creator(is_disagg=is_disagg, target=target, draft=draft, tp_size=4)
+    creator._mapping.enable_attention_dp = True
+
+    assert creator._dflash_kv_layout(target, torch.bfloat16)[0] == 8
+    assert (
+        creator._dflash_kv_layout(_model_config(kv_heads=2, head_dim=128), torch.bfloat16)[0] == 2
+    )
+    assert creator._should_unify_dflash_kv_cache()
+
+
+@pytest.mark.parametrize("is_disagg", [False, True])
+@pytest.mark.parametrize("per_layer_side", ["target", "draft"])
+def test_per_layer_kv_heads_keep_separate_cache(is_disagg, per_layer_side):
+    uniform = _model_config(kv_heads=8, head_dim=128)
+    per_layer = _model_config(kv_heads=[8, 4], head_dim=128)
+    target, draft = (per_layer, uniform) if per_layer_side == "target" else (uniform, per_layer)
+
+    creator = _creator(is_disagg=is_disagg, target=target, draft=draft)
+
+    assert not creator._should_unify_dflash_kv_cache()
+    assert creator._build_unified_dflash_layer_specs(kv_cache_config=None) == ()
+
+
+class _StopLauncher(Exception):
+    pass
+
+
+def _launch_from_disagg_config(monkeypatch, tmp_path, context_servers, generation_servers):
+    """Run the real disaggregated_mpi_worker up to the rank-communicator split."""
+    from tensorrt_llm.commands import serve
+
+    config = tmp_path / "disagg.yaml"
+    config.write_text(
+        yaml.safe_dump(
+            {
+                "context_servers": {"num_instances": 1, **context_servers},
+                "generation_servers": {"num_instances": 1, **generation_servers},
+            }
+        )
+    )
+
+    def stop(*args, **kwargs):
+        raise _StopLauncher
+
+    monkeypatch.setattr(
+        serve._command_telemetry, "apply_disaggregated_telemetry_config", lambda *a, **k: None
+    )
+    monkeypatch.setattr(serve, "set_cuda_device", lambda: None)
+    monkeypatch.setattr("tensorrt_llm.llmapi.disagg_utils.split_world_comm", stop)
+    monkeypatch.delenv("TLLM_DISAGG_RUN_REMOTE_MPI_SESSION_CLIENT", raising=False)
+    with pytest.raises(_StopLauncher):
+        serve.disaggregated_mpi_worker.callback(
+            config_file=str(config), log_level="info", telemetry=False
+        )
+
+
+def test_dflash_on_generation_worker_only_keeps_separate_cache(monkeypatch, tmp_path):
+    _launch_from_disagg_config(
+        monkeypatch,
+        tmp_path,
+        context_servers={},
+        generation_servers={
+            "speculative_config": {"decoding_type": "DFlash", "attention_backend": "TRTLLM"}
+        },
+    )
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=8, head_dim=128)
+
+    for _role in ("context", "generation"):
+        creator = _creator(is_disagg=True, target=target, draft=draft)
+        assert not creator._should_unify_dflash_kv_cache()
+
+
+def test_dflash_with_different_tp_keeps_separate_cache(monkeypatch, tmp_path):
+    spec = {"decoding_type": "DFlash", "attention_backend": "TRTLLM"}
+    _launch_from_disagg_config(
+        monkeypatch,
+        tmp_path,
+        context_servers={"tensor_parallel_size": 4, "speculative_config": spec},
+        generation_servers={"tensor_parallel_size": 8, "speculative_config": spec},
+    )
+    target = _model_config(kv_heads=8, head_dim=128)
+    draft = _model_config(kv_heads=4, head_dim=128)
+
+    for tp_size in (4, 8):
+        creator = _creator(is_disagg=True, target=target, draft=draft, tp_size=tp_size)
+        assert not creator._should_unify_dflash_kv_cache()
+
+
+@pytest.mark.parametrize(
+    "context_backend,generation_backend,shared",
+    [
+        ("TRTLLM", "TRTLLM", True),
+        ("VANILLA", "TRTLLM", False),
+        ("TRTLLM", "VANILLA", False),
+        ("FA4", "TRTLLM", False),
+        ("TRTLLM", "FA4", False),
+        (None, "TRTLLM", False),
+        ("TRTLLM", None, False),
+    ],
+)
+def test_launcher_requires_trtllm_draft_attention_on_both_peers(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    context_backend: str | None,
+    generation_backend: str | None,
+    shared: bool,
+) -> None:
+    specs = []
+    for backend in (context_backend, generation_backend):
+        spec = {"decoding_type": "DFlash"}
+        if backend is not None:
+            spec["attention_backend"] = backend
+        specs.append(spec)
+    _launch_from_disagg_config(
+        monkeypatch,
+        tmp_path,
+        context_servers={"speculative_config": specs[0]},
+        generation_servers={"speculative_config": specs[1]},
+    )
+
+    assert os.environ[_SHARED_DRAFT_KV_ENV] == ("1" if shared else "0")
+    for backend in (context_backend, generation_backend):
+        creator = _creator(
+            is_disagg=True,
+            target=_model_config(kv_heads=8, head_dim=128),
+            draft=_model_config(kv_heads=8, head_dim=128),
+        )
+        creator._speculative_config.attention_backend = backend or "VANILLA"
+        assert creator._should_unify_dflash_kv_cache() is shared
+
+
+@pytest.mark.parametrize("peer_index", [0, 1], ids=["context", "generation"])
+@pytest.mark.parametrize(
+    "server_overrides,spec_overrides,shared",
+    [
+        ({"attn_backend": "VANILLA"}, {}, False),
+        ({}, {"draft_len_schedule": {8: 15}}, False),
+        ({}, {"max_concurrency": 8}, False),
+        ({"kv_cache_config": {"tokens_per_block": 128}}, {}, False),
+        ({"pipeline_parallel_size": 2}, {}, False),
+        ({"context_parallel_size": 2}, {}, False),
+        (
+            {
+                "kv_cache_compression_config": {
+                    "algorithm": "quantization_for_cold_page",
+                    "quant": "nvfp4",
+                }
+            },
+            {},
+            False,
+        ),
+        (
+            {"kv_cache_config": {"dtype": "mixed_nvfp4_fp8", "mixed_precision_last_n": 64}},
+            {},
+            False,
+        ),
+        (
+            {"kv_cache_config": {"dtype": "MIXED_NVFP4_BFLOAT16", "mixed_precision_last_n": 64}},
+            {},
+            False,
+        ),
+        ({"kv_cache_config": {"tokens_per_block": 16}}, {}, True),
+        ({"kv_cache_config": {"tokens_per_block": 64}}, {}, True),
+        ({"kv_cache_config": {"dtype": "fp8"}}, {}, True),
+        ({"kv_cache_config": {"dtype": "bfloat16"}}, {}, True),
+        (
+            {
+                "pipeline_parallel_size": 1,
+                "context_parallel_size": 1,
+                "kv_cache_compression_config": None,
+            },
+            {},
+            True,
+        ),
+        ({}, {"draft_len_schedule": None, "max_concurrency": None}, True),
+    ],
+    ids=[
+        "target-backend",
+        "schedule",
+        "concurrency",
+        "page-128",
+        "pipeline-parallel",
+        "context-parallel",
+        "compression",
+        "mixed-kv",
+        "mixed-kv-uppercase",
+        "page-16",
+        "page-64",
+        "fp8-kv",
+        "bf16-kv",
+        "explicit-defaults",
+        "nulls",
+    ],
+)
+def test_launcher_checks_all_peer_configs(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    peer_index: int,
+    server_overrides: dict,
+    spec_overrides: dict,
+    shared: bool,
+) -> None:
+    servers = [
+        {
+            "speculative_config": {
+                "decoding_type": "DFlash",
+                "attention_backend": "TRTLLM",
+                "max_draft_len": 15,
+            }
+        }
+        for _ in range(2)
+    ]
+    servers[peer_index].update(server_overrides)
+    servers[peer_index]["speculative_config"].update(spec_overrides)
+    _launch_from_disagg_config(monkeypatch, tmp_path, servers[0], servers[1])
+
+    assert os.environ[_SHARED_DRAFT_KV_ENV] == ("1" if shared else "0")
+    # The unchanged, locally eligible peer must follow the common decision.
+    creator = _creator(
+        is_disagg=True,
+        target=_model_config(kv_heads=8, head_dim=128),
+        draft=_model_config(kv_heads=8, head_dim=128),
+    )
+    assert creator._should_unify_dflash_kv_cache() is shared
diff --git a/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_budget_split.py b/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_budget_split.py
index e3ed6bf727c7..d4ffd885e5b3 100644
--- a/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_budget_split.py
+++ b/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_budget_split.py
@@ -332,6 +332,7 @@ def get_num_attention_layers(self) -> int:
         # covered in test_kv_cache_estimation.py; this test is about the windows.
         target_kv_config = KvCacheConfig(enable_block_reuse=False)
         mode = Mock()
+        mode.is_dflash.return_value = False
         mode.is_external_drafter.return_value = False
         # A Mock's truthy return would send get_num_extra_kv_tokens down the
         # one-engine arm; this test's config is a drafter outside the engine.
diff --git a/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_estimation.py b/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_estimation.py
index 5c06f0766e20..291eff2ff3d5 100644
--- a/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_estimation.py
+++ b/tests/unittest/_torch/pyexecutor/kv_cache/test_kv_cache_estimation.py
@@ -34,11 +34,13 @@
 from tensorrt_llm._torch.pyexecutor.config_utils import get_layer_attention_window
 from tensorrt_llm._torch.pyexecutor.kv_cache.kv_cache_manager_v2 import (
     KVCacheLayerOwner,
-    KVCacheLayerSpec,
     KVCacheManagerV2,
     Role,
     _get_static_cache_size_layer_components,
 )
+from tensorrt_llm._torch.pyexecutor.kv_cache.kv_cache_manager_v2 import (
+    KVCacheLayerSpec as KVCacheStorageLayerSpec,
+)
 from tensorrt_llm._torch.pyexecutor.model_engine import ModelEngine
 from tensorrt_llm._torch.pyexecutor.resource_manager import KVCacheManager
 from tensorrt_llm._torch.speculative.interface import SpeculativeDecodingMode
@@ -1250,7 +1252,7 @@ def get_cache_size_per_token(model_config, mapping, **kwargs):
 
 
 def test_v2_layer_spec_costing_preserves_sink_and_reuse_policy() -> None:
-    windowed = KVCacheLayerSpec(
+    windowed = KVCacheStorageLayerSpec(
         global_layer_id=0,
         owner=KVCacheLayerOwner.TARGET,
         dtype=DataType.BF16,
@@ -1908,6 +1910,58 @@ def _mixed_schedule_pretrained(layer_types: list[str], **overrides: object) -> S
     return SimpleNamespace(**fields)
 
 
+@pytest.mark.parametrize("publish_model_geometry", [False, True])
+def test_create_kv_cache_manager_preserves_storage_layer_specs(
+    publish_model_geometry: bool,
+) -> None:
+    captured = {}
+
+    class RecordingManager(KVCacheManagerV2):
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            captured.update(kwargs)
+
+    target = KVCacheStorageLayerSpec(
+        global_layer_id=0,
+        owner=KVCacheLayerOwner.TARGET,
+        dtype=DataType.FP8,
+        num_kv_heads=8,
+        head_dim=128,
+        buffer_roles=(Role.KEY, Role.VALUE),
+    )
+    storage_specs = (
+        target,
+        replace(target, global_layer_id=1, owner=KVCacheLayerOwner.DRAFT, dtype=DataType.BF16),
+    )
+    model_config = SimpleNamespace(
+        pretrained_config=_mixed_schedule_pretrained([_FULL]),
+        quant_config=None,
+        kv_cache_layer_specs=(
+            (ModelKVCacheLayerSpec(head_dim=128, num_kv_heads=8, num_q_heads=8),)
+            if publish_model_geometry
+            else None
+        ),
+    )
+    _create_kv_cache_manager(
+        model_engine=None,
+        kv_cache_manager_cls=RecordingManager,
+        mapping=Mapping(),
+        kv_cache_config=KvCacheConfig(),
+        tokens_per_block=32,
+        max_seq_len=4096,
+        max_batch_size=4,
+        spec_config=None,
+        sparse_attention_config=None,
+        max_num_tokens=1024,
+        max_beam_width=1,
+        kv_connector_manager=None,
+        model_config=model_config,
+        dtype=torch.bfloat16,
+        layer_specs=storage_specs,
+    )
+
+    assert captured["layer_specs"] is storage_specs
+
+
 def _eagle3_one_model_spec_config(num_draft_hidden_layers: int | None = None) -> SimpleNamespace:
     """The fields `get_num_spec_layers` and `should_use_separate_draft_kv_cache`
     read from an Eagle3 one-model speculative config."""
diff --git a/tests/unittest/_torch/pyexecutor/kv_cache/test_kvcm2_integration.py b/tests/unittest/_torch/pyexecutor/kv_cache/test_kvcm2_integration.py
index 83bd77f0c2f4..088d22da8751 100644
--- a/tests/unittest/_torch/pyexecutor/kv_cache/test_kvcm2_integration.py
+++ b/tests/unittest/_torch/pyexecutor/kv_cache/test_kvcm2_integration.py
@@ -766,6 +766,85 @@ def test_mixed_role_coalesced_pool_preserves_slot_stride(kv_first: bool, kv_layo
         manager.shutdown()
 
 
+def test_dflash_block_tables_with_fp8_target_and_bf16_draft() -> None:
+    if not torch.cuda.is_available():
+        pytest.skip("requires CUDA")
+    from tensorrt_llm._torch.speculative.dflash import DFlashWorker
+
+    torch.cuda.init()
+    target = KVCacheLayerSpec(
+        global_layer_id=0,
+        owner=KVCacheLayerOwner.TARGET,
+        dtype=DataType.FP8,
+        num_kv_heads=2,
+        head_dim=8,
+        buffer_roles=(Role.KEY, Role.VALUE),
+    )
+    # Equal byte sizes coalesce the FP8 target and BF16 draft pages, placing
+    # the draft's K buffer at a nonzero offset within each physical slot.
+    draft = replace(
+        target,
+        global_layer_id=1,
+        owner=KVCacheLayerOwner.DRAFT,
+        dtype=DataType.BF16,
+        num_kv_heads=1,
+    )
+    manager = KVCacheManagerV2(
+        KvCacheConfig(
+            max_gpu_total_bytes=16 << 20,
+            host_cache_size=0,
+            enable_block_reuse=False,
+            enable_swa_scratch_reuse=False,
+        ),
+        CacheType.SELF,
+        num_layers=1,
+        num_kv_heads=target.num_kv_heads,
+        head_dim=target.head_dim,
+        tokens_per_block=TOKENS_PER_BLOCK,
+        max_seq_len=MAX_SEQ_LEN,
+        max_batch_size=2,
+        mapping=Mapping(world_size=1, rank=0, tp_size=1, pp_size=1),
+        dtype=DataType.FP8,
+        layer_specs=[target, draft],
+    )
+    try:
+        assert manager._use_per_layer_page_tables
+        request_ids = [7, 9]
+        for request_id, num_blocks in zip(request_ids, (2, 3)):
+            cache = manager._create_kv_cache(request_id, None, None)
+            assert cache.resume(manager._stream.cuda_stream)
+            assert cache.resize(num_blocks * TOKENS_PER_BLOCK, 0)
+
+        worker = object.__new__(DFlashWorker)
+        worker._ctx_unified_cache = True
+        pool = worker._managed_ctx_pool(
+            manager, 1, draft.num_kv_heads, draft.head_dim, torch.bfloat16, 2
+        )
+        assert pool is not None
+        assert worker._init_ctx_block_tables(
+            manager, pool, len(request_ids), 1, draft.num_kv_heads, draft.head_dim, TOKENS_PER_BLOCK
+        )
+        assert worker._ctx_block_offset > 0
+
+        block_offsets = torch.empty(
+            (manager.num_attention_op_pools, len(request_ids), 2, manager.max_blocks_per_seq),
+            dtype=torch.int32,
+            device="cuda",
+        )
+        manager.copy_batch_block_offsets(
+            block_offsets, request_ids, beam_width=1, num_contexts=0, num_seqs=len(request_ids)
+        )
+        metadata = SimpleNamespace(kv_cache_block_offsets=block_offsets)
+        assert worker._refresh_ctx_block_tables(metadata, len(request_ids))
+
+        expected = manager.get_batch_cache_indices(request_ids, layer_idx=draft.global_layer_id)
+        for row, indices in enumerate(expected):
+            assert len(indices) == (2, 3)[row]
+            assert worker._ctx_block_tables[row, : len(indices)].tolist() == indices
+    finally:
+        manager.shutdown()
+
+
 def test_nvfp4_scale_indices_with_mixed_bf16_storage() -> None:
     if not torch.cuda.is_available():
         pytest.skip("requires CUDA")
diff --git a/tests/unittest/_torch/speculative/hw_agnostic/test_dflash_copy_skip.py b/tests/unittest/_torch/speculative/hw_agnostic/test_dflash_copy_skip.py
index f88ba9687763..628897450b31 100644
--- a/tests/unittest/_torch/speculative/hw_agnostic/test_dflash_copy_skip.py
+++ b/tests/unittest/_torch/speculative/hw_agnostic/test_dflash_copy_skip.py
@@ -25,8 +25,8 @@ def _worker(req_to_slot):
         _batch_to_slot_buffer=DeviceCopyBuffer(torch.zeros(8, dtype=torch.long)),
         _write_ctx_len=lambda updates: None,
     )
-    # Generation requests are already assigned in these tests; prepare assigns nothing.
-    worker._assign_slot = lambda rid: worker._req_to_slot.get(rid)
+    # Slot allocation is outside these copy tests; missing IDs use the dummy slot.
+    worker._assign_slot = lambda rid, updates=None: worker._req_to_slot.get(rid)
     return worker
 
 
diff --git a/tests/unittest/_torch/speculative/test_spec_tail_kernels.py b/tests/unittest/_torch/speculative/test_spec_tail_kernels.py
index ffddda1808bc..3681fbe28fe1 100644
--- a/tests/unittest/_torch/speculative/test_spec_tail_kernels.py
+++ b/tests/unittest/_torch/speculative/test_spec_tail_kernels.py
@@ -64,7 +64,8 @@ def test_strict_accept_matches_eager(num_ctx, num_gens, draft_len):
     assert torch.equal(accepted[num_ctx:], ref_accepted[num_ctx:])
 
 
-def test_decode_draft_block_table_matches_eager():
+@pytest.mark.parametrize("block_offset", [0, 3])
+def test_decode_draft_block_table_matches_eager(block_offset: int) -> None:
     num_pools, rows, n_blocks, divisor = 2, 9, 3000, 6
     gen = torch.Generator(device="cuda").manual_seed(1)
     src = torch.randint(
@@ -77,16 +78,21 @@ def test_decode_draft_block_table_matches_eager():
     view = src[1, :rows, 0]
     view[0] = -torch.randint(1, 1 << 20, (n_blocks,), device="cuda", generator=gen).to(torch.int32)
     view[1] = 0
+    view[2] = block_offset
     assert bool((view < 0).float().mean() > 0.3) and bool((view == 0).float().mean() > 0.2)
     table = torch.full((rows + 3, n_blocks), -7, dtype=torch.int32, device="cuda")
     counts = torch.full((rows + 3,), -7, dtype=torch.long, device="cuda")
-    decode_draft_block_table(view, divisor, table, counts)
-    encoded = view.to(torch.int64)
+    original = src.clone()
+    decode_draft_block_table(view, divisor, table, counts, block_offset=block_offset)
+    assert torch.equal(src, original)
+    encoded = view.to(torch.int64) - block_offset
     assert torch.equal(counts[:rows], (encoded >= 0).sum(dim=1))
     expected = encoded.clamp(min=0).div(divisor, rounding_mode="floor").to(torch.int32)
     assert torch.equal(table[:rows], expected)
     assert counts[0].item() == 0 and bool((table[0] == 0).all())
-    assert counts[1].item() == n_blocks and bool((table[1] == 0).all())
+    assert counts[1].item() == (n_blocks if block_offset == 0 else 0)
+    assert bool((table[1] == 0).all())
+    assert counts[2].item() == n_blocks and bool((table[2] == 0).all())
     assert bool((table[rows:] == -7).all()) and bool((counts[rows:] == -7).all())
 
 
diff --git a/tests/unittest/disaggregated/test_peer.py b/tests/unittest/disaggregated/test_peer.py
index 15670e3de0c6..fb783480de80 100644
--- a/tests/unittest/disaggregated/test_peer.py
+++ b/tests/unittest/disaggregated/test_peer.py
@@ -1445,3 +1445,29 @@ def test_attention_replicated_validation_rejects_missing_index_key_view():
     peer_pt = make_page_table(global_layer_ids=[0])
     with pytest.raises(ValueError, match="AttentionPolicy.*differ on overlapping layers"):
         AttentionPolicy.validate_peer_compatible(self_pt, peer_pt)
+
+
+@pytest.mark.parametrize(("local_draft", "peer_draft"), [(0, 0), (1, 1), (1, 0), (0, 1)])
+def test_peer_registrar_requires_matching_unified_draft_layers(
+    local_draft: int, peer_draft: int
+) -> None:
+    self_rankinfo = make_rankinfo(instance_name="local")
+    self_rankinfo.num_draft_layers = local_draft
+    reg = _make_peer_registrar(self_rankinfo)
+    peer_ri = make_rankinfo(
+        instance_name="peer",
+        instance_rank=1,
+        layer_num_per_pp=[2],
+        page_table=make_page_table(),
+    )
+    peer_ri.num_draft_layers = peer_draft
+    if local_draft != peer_draft:
+        with pytest.raises(
+            ValueError,
+            match=f"draft KV layer count differs.*local={local_draft}, peer={peer_draft}",
+        ):
+            reg.register(peer_ri.instance_name, peer_ri.instance_rank, peer_ri)
+        return
+
+    reg.register(peer_ri.instance_name, peer_ri.instance_rank, peer_ri)
+    assert reg.get_peer_rank_info("peer", 1) == peer_ri
diff --git a/tests/unittest/disaggregated/test_rank_info.py b/tests/unittest/disaggregated/test_rank_info.py
index a1d25e01c3f3..a9ef379c93e7 100644
--- a/tests/unittest/disaggregated/test_rank_info.py
+++ b/tests/unittest/disaggregated/test_rank_info.py
@@ -15,6 +15,7 @@
 
 from types import SimpleNamespace
 
+import msgpack
 import numpy as np
 import pytest
 
@@ -203,3 +204,64 @@ def test_rank_info_represents_cache_element_bytes(
     restored = RankInfo.from_bytes(rank_info.to_bytes())
     assert restored.attention.element_bytes == expected_element_bytes
     assert isinstance(restored.attention.element_bytes, expected_type)
+
+
+@pytest.mark.parametrize(
+    ("layer_specs", "draft_layer_ids", "expected"),
+    [
+        (None, (), 0),
+        # Folded MTP/Eagle draft layers have no layer specs and travel as
+        # ordinary layers, so they are not advertised.
+        (None, (36,), 0),
+        ((object(),), (36, 37), 2),
+    ],
+)
+def test_from_kv_cache_manager_reports_unified_draft_layers(
+    monkeypatch, layer_specs, draft_layer_ids, expected
+) -> None:
+    monkeypatch.setattr(rank_info_module, "build_page_table_from_manager", lambda _: None)
+    manager = SimpleNamespace(
+        mapping=SimpleNamespace(
+            rank=0,
+            tp_size=1,
+            tp_rank=0,
+            pp_size=1,
+            pp_rank=0,
+            dp_size=1,
+            cp_size=1,
+            cp_rank=0,
+            enable_attention_dp=False,
+        ),
+        num_kv_heads_per_layer=[8, 8],
+        pp_layers=[0, 1],
+        tokens_per_block=32,
+        head_dim=128,
+        dtype=bindings.DataType.HALF,
+        kv_factor=2,
+        layer_specs=layer_specs,
+        draft_layer_ids=draft_layer_ids,
+    )
+
+    info = RankInfo.from_kv_cache_manager("ctx", manager, device_id=0)
+
+    assert info.num_draft_layers == expected
+    assert RankInfo.from_bytes(info.to_bytes()).num_draft_layers == expected
+
+
+def test_rank_info_from_bytes_defaults_missing_draft_layer_count() -> None:
+    ri = RankInfo(
+        instance_name="gen_0",
+        instance_rank=0,
+        tp_size=1,
+        tp_rank=0,
+        pp_size=1,
+        pp_rank=0,
+        layer_num_per_pp=[32],
+        sender_endpoints=[],
+        self_endpoint="",
+        transfer_engine_info=b"",
+    )
+    payload = msgpack.unpackb(ri.to_bytes(), strict_map_key=False)
+    del payload["num_draft_layers"]
+
+    assert RankInfo.from_bytes(msgpack.packb(payload)).num_draft_layers == 0