diff --git a/tensorrt_llm/_torch/pyexecutor/_util.py b/tensorrt_llm/_torch/pyexecutor/_util.py
index 7e31ad99cff7..76014baa531a 100644
--- a/tensorrt_llm/_torch/pyexecutor/_util.py
+++ b/tensorrt_llm/_torch/pyexecutor/_util.py
@@ -642,6 +642,7 @@ def __init__(
         self._dummy_encoder_inputs: List[MultimodalParams] = []
         self._profiling_stage_data = profiling_stage_data
         self._is_disagg = is_disagg
+        self._disable_overlap_scheduler = llm_args.disable_overlap_scheduler
         self._cache_transceiver_config = llm_args.cache_transceiver_config
         self._execution_stream = execution_stream
         self._kv_cache_manager_cls = self._get_model_kv_cache_manager_cls(
@@ -1462,6 +1463,7 @@ def _create_kv_cache_manager(
             execution_stream=self._execution_stream,
             layer_mask=spec_dec_layer_mask,
             is_disagg=self._is_disagg,
+            disable_overlap_scheduler=self._disable_overlap_scheduler,
             cold_page_codec_provider=cold_page_codec_provider,
             joint_kv_cache_reuse=self._joint_kv_cache_reuse,
         )
@@ -1668,6 +1670,7 @@ def _create_one_model_draft_kv_cache_manager(
             layer_mask=spec_dec_layer_mask,
             num_layers=num_draft_layers,
             is_disagg=self._is_disagg,
+            disable_overlap_scheduler=self._disable_overlap_scheduler,
             cold_page_codec_provider=cold_page_codec_provider,
             joint_kv_cache_reuse=self._joint_kv_cache_reuse,
         )
@@ -2044,6 +2047,7 @@ def _create_cross_kv_cache_manager(
             num_layers=num_layers,
             num_kv_heads=num_kv_heads,
             head_dim=head_dim,
+            disable_overlap_scheduler=self._disable_overlap_scheduler,
             kv_cache_type=tensorrt_llm.bindings.internal.batch_manager.
             CacheType.CROSS,
         )
@@ -2363,6 +2367,7 @@ def _create_kv_cache_manager(
         head_dim: Optional[int] = None,
         kv_cache_type=None,
         is_disagg: bool = False,
+        disable_overlap_scheduler: bool = False,
         cold_page_codec_provider: Optional[object] = None,
         joint_kv_cache_reuse: bool = False) -> KVCacheManager:
     """
@@ -2504,6 +2509,8 @@ def _create_kv_cache_manager(
         manager_extra_kwargs[
             "cold_page_codec_provider"] = cold_page_codec_provider
         manager_extra_kwargs["joint_kv_cache_reuse"] = joint_kv_cache_reuse
+        manager_extra_kwargs[
+            "disable_overlap_scheduler"] = disable_overlap_scheduler
     if issubclass(kv_cache_manager_cls, MambaHybridCacheManagerV2):
         manager_extra_kwargs["is_disagg"] = is_disagg
 
@@ -3025,26 +3032,64 @@ def create_kv_cache_compression_manager(
     return None
 
 
+def is_disagg_enabled(cache_transceiver_config) -> bool:
+    """True when a cache transceiver backend is configured."""
+    return (cache_transceiver_config is not None
+            and cache_transceiver_config.backend is not None)
+
+
 def compute_max_num_sequences(mapping: Mapping,
                               max_batch_size: int,
                               disable_overlap_scheduler: bool,
                               enable_overlap_headroom: bool = False) -> int:
     """Size the sequence-slot pool (and the sampler state it indexes).
 
-    ``enable_overlap_headroom`` is intentionally opt-in. Disaggregated
-    attention-DP needs a second non-PP slot set because the V2 scheduler can
-    backfill seats before the overlap scheduler releases the previous
-    iteration's terminal slots. Pipeline parallelism already sizes the pool
-    by ``pp_size``.
+    ``enable_overlap_headroom`` is intentionally opt-in; see
+    ``should_enable_disagg_adp_overlap_headroom`` for when it is set. It buys one
+    extra micro-batch worth of slots, because a finished request's teardown is
+    deferred by one iteration and its slot is still held while the replacement
+    batch is admitted (nvbug 6627795). Pipeline parallelism already sizes the
+    pool by ``pp_size``.
     """
     if mapping.has_pp():
         num_micro_batches = mapping.pp_size
     else:
-        num_micro_batches = (2 if enable_overlap_headroom
-                             and not disable_overlap_scheduler else 1)
+        num_micro_batches = (2 if enable_overlap_headroom else 1)
     return max_batch_size * num_micro_batches
 
 
+def resolve_max_num_sequences(model_engine,
+                              mapping: Mapping,
+                              max_batch_size: int,
+                              llm_args,
+                              max_num_sequences: Optional[int] = None) -> int:
+    """Resolve the seat-pool size for a consumer, without re-deriving it.
+
+    Order of preference, and the order matters:
+
+    1. an explicitly supplied value -- the caller already has the number the
+       engine published;
+    2. ``model_engine.max_num_seq_slots`` -- the engine's own pool, which is
+       what every seat-keyed pool was sized against;
+    3. only then a fresh ``compute_max_num_sequences``, reusing the engine's
+       headroom gate so the fallback cannot size the pool below the engine's.
+    """
+    if max_num_sequences is not None:
+        return max_num_sequences
+    engine_seats = getattr(model_engine, "max_num_seq_slots", None)
+    if engine_seats is not None:
+        return engine_seats
+    # Engines that predate the attribute (unit-test stubs, mm-encoder-only
+    # engines): recompute, but with the same gate the engine would have used.
+    return compute_max_num_sequences(mapping,
+                                     max_batch_size,
+                                     llm_args.disable_overlap_scheduler,
+                                     enable_overlap_headroom=getattr(
+                                         model_engine,
+                                         "_enable_disagg_adp_overlap_headroom",
+                                         False))
+
+
 def should_enable_adp_dummy_fixes(mapping: Mapping) -> bool:
     """Enable transactional ADP dummy handling while PP remains follow-up."""
     return not mapping.has_pp()
@@ -3073,11 +3118,10 @@ def should_enable_disagg_adp_overlap_headroom(
         mapping: Mapping,
         cache_transceiver_config: Optional[CacheTransceiverConfig],
         disable_overlap_scheduler: bool) -> bool:
-    """Gate extra sequence slots to non-PP disaggregated attention-DP."""
-    is_disagg = (cache_transceiver_config is not None
-                 and cache_transceiver_config.backend is not None)
-    return (mapping.enable_attention_dp and is_disagg and not mapping.has_pp()
-            and not disable_overlap_scheduler)
+    """Gate extra sequence slots to non-PP attention DP."""
+    is_disagg = is_disagg_enabled(cache_transceiver_config)
+    return (mapping.enable_attention_dp and not mapping.has_pp()
+            and (is_disagg or not disable_overlap_scheduler))
 
 
 def create_py_executor_instance(
@@ -3115,15 +3159,18 @@ def create_py_executor_instance(
 
     spec_config = model_engine.spec_config
 
-    if max_num_sequences is None:
-        max_num_sequences = compute_max_num_sequences(
-            mapping, max_batch_size, llm_args.disable_overlap_scheduler)
+    is_disagg = is_disagg_enabled(cache_transceiver_config)
+
+    max_num_sequences = resolve_max_num_sequences(
+        model_engine,
+        mapping,
+        max_batch_size,
+        llm_args,
+        max_num_sequences=max_num_sequences)
 
     logger.info(
         f"max_seq_len={max_seq_len}, max_num_requests={max_num_sequences}, max_num_tokens={max_num_tokens}, max_batch_size={max_batch_size}"
     )
-    is_disagg = (cache_transceiver_config is not None
-                 and cache_transceiver_config.backend is not None)
     for key, value in llm_args.extra_resource_managers.items():
         if key in resources:
             raise ValueError(
@@ -3312,8 +3359,10 @@ def create_py_executor_instance(
 
     # When scheduler_capacity == 1, attention dp dummy request will prevent the scheduling of DISAGG_GENERATION_INIT.
     # Enlarge scheduler capacity to avoid DISAGG_GENERATION_INIT stuck in the scheduler.
-    # V1 scheduler handles overlap via two_step_lookahead, so skip the
-    # slot-pool overlap factor here.
+    # V1 scheduler handles overlap via two_step_lookahead, so the capacity
+    # scheduler's budget stays at the pipeline-depth bound and deliberately does
+    # not follow the sequence-slot pool: the overlap headroom is spare seats for
+    # leases already held, not extra admission.
     scheduler_capacity = max_batch_size * mapping.pp_size
     if scheduler_capacity == 1 and mapping.enable_attention_dp and kv_cache_manager:
         scheduler_capacity += 1
@@ -3500,22 +3549,22 @@ def create_py_executor_instance(
 
 
 def create_torch_sampler_args(
-    mapping: Mapping,
     *,
     max_seq_len: int,
-    max_batch_size: int,
     speculative_config: SpeculativeConfig,
     max_beam_width: int,
     disable_overlap_scheduler: bool,
     enable_async_worker: bool,
     enable_speculative_beam_history_d2h: bool,
-    max_num_sequences: Optional[int] = None,
+    max_num_sequences: int,
 ):
-    # The sampler's per-slot state is indexed by sequence slots, so it must
-    # be sized identically to the executor's slot pool.
-    if max_num_sequences is None:
-        max_num_sequences = compute_max_num_sequences(
-            mapping, max_batch_size, disable_overlap_scheduler)
+    # The sampler's per-slot state is indexed by sequence slots, so it must be
+    # sized identically to the executor's slot pool. `max_num_sequences` is
+    # required, not optional: the old default recomputed the pool from
+    # `mapping`/`max_batch_size` *without* the overlap-headroom gate, so it could
+    # only ever produce a smaller number than the slots it indexes. Those two
+    # parameters are gone with it -- keeping them would leave the raw material
+    # for the same re-derivation lying next to the resolved value.
     max_draft_len = (0 if speculative_config is None else
                      speculative_config.max_draft_len)
     max_total_draft_tokens = (0 if speculative_config is None else
@@ -3547,10 +3596,15 @@ def instantiate_sampler(
     enable_async_worker = (confidential_compute_enabled()
                            or llm_args.sampler_force_async_worker)
 
-    sampler_args = create_torch_sampler_args(
+    max_num_sequences = resolve_max_num_sequences(
+        engine,
         mapping,
+        max_batch_size,
+        llm_args,
+        max_num_sequences=max_num_sequences)
+
+    sampler_args = create_torch_sampler_args(
         max_seq_len=engine.max_seq_len,
-        max_batch_size=max_batch_size,
         speculative_config=speculative_config,
         max_beam_width=max_beam_width,
         disable_overlap_scheduler=llm_args.disable_overlap_scheduler,
diff --git a/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py b/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py
index 97d117448bb9..c9caafd4302a 100644
--- a/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py
+++ b/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py
@@ -931,6 +931,7 @@ def __init__(
         is_disagg: bool = False,
         enable_stats: bool = False,
         num_reserved_index_slots: int = 1,
+        disable_overlap_scheduler: bool = False,
         is_estimating_kv_cache: bool = False,
         cold_page_codec_provider: Optional[object] = None,
         joint_kv_cache_reuse: bool = False,
@@ -1452,14 +1453,18 @@ def create_cold_page_codec(cache_config: object) -> Optional[object]:
         # (TRANS_IN_PROGRESS) and continue to hold their index slots. The 2x
         # capacity lets the next batch of active requests acquire slots without
         # waiting for the previous batch's transfers to finish.
+        needs_extra_index_slots = is_disagg or (
+            mapping.enable_attention_dp and not disable_overlap_scheduler and not mapping.has_pp()
+        )
         max_num_sequences = max_batch_size * mapping.pp_size
         assert num_reserved_index_slots >= 0, "num_reserved_index_slots must be non-negative"
         index_mapper_capacity = (
-            max_num_sequences * (2 if is_disagg else 1) + num_reserved_index_slots
+            max_num_sequences * (2 if needs_extra_index_slots else 1) + num_reserved_index_slots
         )
         logger.info(
             f"KVCacheManagerV2: IndexMapper capacity={index_mapper_capacity} "
             f"(max_num_sequences={max_num_sequences}, is_disagg={is_disagg}, "
+            f"disable_overlap_scheduler={disable_overlap_scheduler}, "
             f"num_reserved_index_slots={num_reserved_index_slots}, "
             f"max_beam_width={max_beam_width})"
         )
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py
index 15a6c5f0a7f1..ec85f029e2f8 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py
@@ -107,7 +107,7 @@
 from .scheduler import (RequestScheduler, ScheduledRequests,
                         SerializableSchedulerOutput, WaitingQueue,
                         create_waiting_queue)
-from .scheduler.adp_router import ADPRouter
+from .scheduler.adp_router import ADPRouter, count_retiring_requests
 
 if TYPE_CHECKING:
     from ray.actor import ActorHandle
@@ -692,6 +692,8 @@ def __init__(
         # can receive the transfer-manager reference at construction time.
         self.adp_router: ADPRouter = ADPRouter.create(
             dist=self.dist,
+            has_seq_slot_headroom=getattr(
+                model_engine, "_enable_disagg_adp_overlap_headroom", False),
             kv_cache_manager=self.kv_cache_manager,
             attention_dp_config=self.llm_args.attention_dp_config,
             async_transfer_manager=self.async_transfer_manager,
@@ -5794,7 +5796,7 @@ def _validate_request(self, request: LlmRequest):
         self._validate_request_budget(request)
 
     def _fetch_and_enqueue_requests(self, waiting_queue: WaitingQueue,
-                                    total_num_active_requests: int) -> None:
+                                    total_num_live_requests: int) -> None:
         """Fetch requests from request_queue and enqueue to waiting_queue."""
         # Block new requests while control requests are pending
         if len(self.control_requests) != 0:
@@ -5805,7 +5807,7 @@ def _fetch_and_enqueue_requests(self, waiting_queue: WaitingQueue,
         # blocking would keep the loop from reaching the
         # `should_stop_processing` check that ends it, deadlocking shutdown()
         # on `shutdown_event`.
-        idle = (total_num_active_requests == 0 and len(waiting_queue) == 0
+        idle = (total_num_live_requests == 0 and len(waiting_queue) == 0
                 and not self.is_shutdown)
         if idle:
             # In Ray path (TLLM_DISABLE_MPI=1), use a periodic heartbeat timeout so rank 0
@@ -6013,14 +6015,16 @@ def _fetch_new_requests(
                 s.num_active_requests for s in all_rank_states
             ]
             total_num_active_requests = sum(all_ranks_num_active_requests)
+            total_num_live_requests = total_num_active_requests + sum(
+                s.num_retiring_requests for s in all_rank_states)
         else:
             total_num_active_requests = len(active_requests)
+            total_num_live_requests = total_num_active_requests
             all_ranks_num_active_requests = None
             all_rank_states = None
 
         # 2. Fetch and enqueue to waiting queue
-        self._fetch_and_enqueue_requests(waiting_queue,
-                                         total_num_active_requests)
+        self._fetch_and_enqueue_requests(waiting_queue, total_num_live_requests)
 
         # 3. Pop requests from waiting queue
         new_requests = self._pop_from_waiting_queue(
@@ -7225,7 +7229,11 @@ def _pad_attention_dp_dummy_request(self):
             return
 
         expected_num_active_requests = self.expected_num_active_requests
-        if expected_num_active_requests < len(self.active_requests):
+        num_routable_active_requests = len(self.active_requests)
+        if self.adp_router.exclude_retiring_requests:
+            num_routable_active_requests -= count_retiring_requests(
+                self.active_requests)
+        if expected_num_active_requests < num_routable_active_requests:
             # Not fatal, and not a capacity violation. The router derives this
             # value as
             #   min(max(ceil(multiplier * fair_share), max(per_rank_loads)),
@@ -7246,11 +7254,12 @@ def _pad_attention_dp_dummy_request(self):
             # event loop on every affected rank at once, leaving the survivors
             # to HangDetector-abort.
             logger.warning(
-                f"active_requests ({len(self.active_requests)}) exceeds "
+                f"routable active_requests "
+                f"({num_routable_active_requests}) exceeds "
                 f"expected_num_active_requests "
                 f"({expected_num_active_requests}); tolerating (a busy rank "
                 f"needs no attention-DP dummy).")
-            expected_num_active_requests = len(self.active_requests)
+            expected_num_active_requests = num_routable_active_requests
 
         num_active_request = self._count_schedulable_active_requests()
 
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor_creator.py b/tensorrt_llm/_torch/pyexecutor/py_executor_creator.py
index 56e6254fe7e5..adfdcd62c7e1 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor_creator.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor_creator.py
@@ -37,8 +37,8 @@
                            get_spec_resource_manager)
 from ..virtual_memory import scope as virtual_memory_scope
 from ._util import (KvCacheCreator, _adjust_torch_mem_fraction,
-                    create_py_executor_instance, instantiate_sampler, is_mla,
-                    validate_feature_combination)
+                    create_py_executor_instance, instantiate_sampler,
+                    is_disagg_enabled, is_mla, validate_feature_combination)
 from .config_utils import (is_hybrid_linear, is_minimax_m3,
                            resolve_cache_transceiver_config,
                            uses_vswa_kv_cache_layout)
@@ -756,15 +756,12 @@ def allocation_scope(current_stage: ExecutorMemoryType):
     if guided_decoding_config is not None:
         with allocation_scope(ExecutorMemoryType.GUIDED_DECODER):
             if mapping.is_last_pp_rank():
-                guided_decoder_slots = (max_num_seq_slots if getattr(
-                    model_engine, "_enable_disagg_adp_overlap_headroom", False)
-                                        else max_batch_size)
                 kwargs = {
                     "guided_decoding_config": guided_decoding_config,
-                    # The disaggregated attention-DP overlap path follows the
-                    # expanded slot pool. Other configurations retain
-                    # max_batch_size.
-                    "max_num_sequences": guided_decoder_slots,
+                    # The guided decoder's state is indexed by py_seq_slot
+                    # (guided_decoder.py: grammar_matchers[req.seq_slot], the
+                    # bitmask rows), so it must span the whole seat pool.
+                    "max_num_sequences": max_num_seq_slots,
                     "vocab_size_padded": model_engine.model.vocab_size_padded,
                     "rank": mapping.rank,
                 }
@@ -875,8 +872,7 @@ def allocation_scope(current_stage: ExecutorMemoryType):
     if model_engine.model.model_config.is_generation:
         #NOTE: non-generation models do not have kv cache
 
-        is_disagg = (cache_transceiver_config is not None
-                     and cache_transceiver_config.backend is not None)
+        is_disagg = is_disagg_enabled(cache_transceiver_config)
         is_hybrid = is_hybrid_linear(
             model_engine.model.model_config.pretrained_config)
 
diff --git a/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py b/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py
index 220136a4b6d3..048543427f68 100644
--- a/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py
+++ b/tensorrt_llm/_torch/pyexecutor/scheduler/adp_router.py
@@ -27,6 +27,8 @@
 
 from tensorrt_llm.logger import logger
 
+from ..llm_request import LlmRequestState
+
 if TYPE_CHECKING:
     from tensorrt_llm._torch.distributed.communicator import Distributed
     from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequest
@@ -47,6 +49,21 @@ def _num_input_tokens(request) -> int:
     return len(getattr(request, "input_token_ids", []))
 
 
+def is_retiring_request(request) -> bool:
+    """True if ``request`` has produced its final token and is being torn down."""
+    return request.state == LlmRequestState.GENERATION_TO_COMPLETE
+
+
+def build_active_requests_for_overlap(active_requests):
+    """Return ``active_requests`` without the requests that are already retiring."""
+    return [req for req in active_requests if not is_retiring_request(req)]
+
+
+def count_retiring_requests(active_requests) -> int:
+    """Count the retiring requests in ``active_requests``."""
+    return sum(1 for req in active_requests if is_retiring_request(req))
+
+
 @dataclass
 class RankIterStatsPayload:
     """Per-rank IterationStats payload piggybacked on the ADP allgather."""
@@ -99,6 +116,7 @@ class RankState:
     rank: int
     num_active_requests: int = 0
     num_active_tokens: int = 0
+    num_retiring_requests: int = 0
     iter_stats: RankIterStatsPayload = field(default_factory=RankIterStatsPayload)
 
     def copy_iter_stats_from(self, iter_stats_payload: RankIterStatsPayload | None) -> None:
@@ -112,6 +130,7 @@ def serialize(self) -> list[int]:
             self.rank,
             self.num_active_requests,
             self.num_active_tokens,
+            self.num_retiring_requests,
             *self.iter_stats.serialize(),
         ]
 
@@ -119,7 +138,7 @@ def serialize(self) -> list[int]:
     def deserialize(cls, data: list[int]) -> RankState:
         """Deserialize from a flat list received via allgather."""
         values = list(data)
-        rank_state_prefix_field_count = 3
+        rank_state_prefix_field_count = 4
         rank_state_fields = fields(cls)[:rank_state_prefix_field_count]
         max_field_count = rank_state_prefix_field_count + len(fields(RankIterStatsPayload))
         if len(values) < 1:
@@ -140,6 +159,7 @@ def deserialize(cls, data: list[int]) -> RankState:
             rank=rank_values[0],
             num_active_requests=rank_values[1],
             num_active_tokens=rank_values[2],
+            num_retiring_requests=rank_values[3],
             iter_stats=RankIterStatsPayload.deserialize(values[rank_state_prefix_field_count:]),
         )
 
@@ -163,13 +183,15 @@ class ADPRouter(ABC):
 
     needs_prefix_matches: bool = False
 
-    def __init__(self, dist: Distributed):
+    def __init__(self, dist: Distributed, has_seq_slot_headroom: bool = True):
         self.dist = dist
+        self.exclude_retiring_requests = has_seq_slot_headroom
 
     @classmethod
     def create(
         cls,
         dist: "Distributed",
+        has_seq_slot_headroom: bool,
         kv_cache_manager=None,
         attention_dp_config=None,
         async_transfer_manager=None,
@@ -178,6 +200,9 @@ def create(
 
         Args:
             dist: Distributed communicator.
+            has_seq_slot_headroom: Whether the executor's sequence-slot pool was
+                sized with the extra overlap headroom. The retiring-request
+                correction is only applied when those extra seats exist.
             kv_cache_manager: KV cache manager instance (may be None).
             attention_dp_config: AttentionDpConfig instance (may be None).
             async_transfer_manager: PyExecutor's AsyncTransferManager, used by
@@ -199,6 +224,7 @@ def create(
             # KV-cache-aware path and takes precedence when both are enabled.
             return ConversationAwareADPRouter(
                 dist=dist,
+                has_seq_slot_headroom=has_seq_slot_headroom,
                 max_sessions=attention_dp_config.kv_cache_routing_max_sessions,
                 fair_share_multiplier=attention_dp_config.kv_cache_routing_fair_share_multiplier,
                 new_conv_placement=attention_dp_config.kv_cache_routing_new_conv_placement,
@@ -212,6 +238,7 @@ def create(
         ):
             return KVCacheAwareADPRouter(
                 dist=dist,
+                has_seq_slot_headroom=has_seq_slot_headroom,
                 kv_cache_manager=kv_cache_manager,
                 load_balance_weight=attention_dp_config.kv_cache_routing_load_balance_weight,
                 match_rate_threshold=attention_dp_config.kv_cache_routing_match_rate_threshold,
@@ -221,7 +248,7 @@ def create(
                 account_for_in_transfer=attention_dp_config.kv_cache_routing_account_for_in_transfer,
             )
 
-        return DefaultADPRouter(dist=dist)
+        return DefaultADPRouter(dist=dist, has_seq_slot_headroom=has_seq_slot_headroom)
 
     @abstractmethod
     def create_rank_state(
@@ -256,7 +283,19 @@ def gather_all_rank_states(
             iter_stats_payload: Completed previous-iteration stats payload to
                 piggyback on this allgather, if one is pending.
         """
-        local_state = self.create_rank_state(active_requests, new_requests or [])
+        # A request whose teardown the overlap scheduler has merely deferred is
+        # not load, and must not hold admission capacity that nothing can spend
+        # (nvbug 6627795).
+        if self.exclude_retiring_requests:
+            active_requests_for_overlap = build_active_requests_for_overlap(active_requests)
+            num_retiring_requests = len(active_requests) - len(active_requests_for_overlap)
+        else:
+            active_requests_for_overlap = active_requests
+            num_retiring_requests = 0
+        local_state = self.create_rank_state(active_requests_for_overlap, new_requests or [])
+        # Reported separately so the executor loop can still tell that these
+        # ranks are not idle and keep its idle-fetch wait collective.
+        local_state.num_retiring_requests = num_retiring_requests
         local_state.copy_iter_stats_from(iter_stats_payload)
         responses = self.dist.tp_allgather(local_state.serialize())
         return [RankState.deserialize(data=resp) for resp in responses]
@@ -507,6 +546,7 @@ def __init__(
         self,
         dist: "Distributed",
         kv_cache_manager,
+        has_seq_slot_headroom: bool = True,
         load_balance_weight: float = 1.0,
         match_rate_threshold: float = 0.1,
         fair_share_multiplier: float = 2.0,
@@ -514,7 +554,7 @@ def __init__(
         async_transfer_manager=None,
         account_for_in_transfer: bool = False,
     ):
-        super().__init__(dist)
+        super().__init__(dist, has_seq_slot_headroom=has_seq_slot_headroom)
         self.kv_cache_manager = kv_cache_manager
         self.load_balance_weight = load_balance_weight
         self.match_rate_threshold = match_rate_threshold
@@ -812,11 +852,12 @@ class ConversationAwareADPRouter(ADPRouter):
     def __init__(
         self,
         dist: "Distributed",
+        has_seq_slot_headroom: bool = True,
         max_sessions: int = DEFAULT_MAX_SESSIONS,
         fair_share_multiplier: float = 2.0,
         new_conv_placement: str = "round_robin",
     ):
-        super().__init__(dist)
+        super().__init__(dist, has_seq_slot_headroom=has_seq_slot_headroom)
         self._conv_to_rank: "OrderedDict[str, int]" = OrderedDict()
         self._max_sessions = max(1, int(max_sessions))
         self._fair_share_multiplier = max(1.0, float(fair_share_multiplier))
@@ -985,8 +1026,8 @@ def _next_rr(soft_cap: int) -> int:
 
         # Sticky returns use the hard cap, so a rank may now exceed the pre-loop
         # soft `expected`. Re-bump so the returned value covers the actual
-        # per-rank max -- _pad_attention_dp_dummy_request asserts
-        # expected >= len(active_requests) on every rank.
+        # per-rank max -- _pad_attention_dp_dummy_request compares `expected`
+        # against each rank's routable active count.
         expected_num_active_requests = max(
             expected_num_active_requests, max(all_ranks_num_active_requests)
         )
diff --git a/tensorrt_llm/_torch/speculative/eagle3.py b/tensorrt_llm/_torch/speculative/eagle3.py
index 7214d2023877..a9ad5b4429d0 100644
--- a/tensorrt_llm/_torch/speculative/eagle3.py
+++ b/tensorrt_llm/_torch/speculative/eagle3.py
@@ -43,7 +43,8 @@ def __init__(self,
                  max_num_requests: int,
                  max_seq_len: int,
                  max_num_tokens: int,
-                 sa_manager=None):
+                 sa_manager=None,
+                 num_seq_slots: Optional[int] = None):
         self.dtype = dtype
         self.max_draft_len = config.max_draft_len
         self.hidden_size = hidden_size
@@ -51,9 +52,17 @@ def __init__(self,
         self.max_seq_len = max_seq_len
         # Optional SA manager for EAGLE3+SA mode
         self.sa_manager = sa_manager
+        # ``slot_manager`` hands out slots keyed by request id and holds them for
+        # the request's whole lifetime, so the pool must span the executor's
+        # sequence-slot pool -- 2 * max_batch_size under the attention-DP overlap
+        # headroom, where a retiring request keeps its slot for one more iteration
+        # while its replacement is admitted (nvbug-6627795). None means no headroom.
+        self.num_seq_slots = max(num_seq_slots or 0, max_num_requests)
         # There could be dummy request for padding batch when using CUDA graph.
         # Reserve one more slot for the dummy request.
-        slot_size = self.max_seq_len + 1
+        # NOTE: max_seq_len is kept as a floor purely to preserve the historical
+        # (over-)sizing; it is a token count, not a slot count.
+        slot_size = max(self.num_seq_slots, self.max_seq_len) + 1
         self.slot_manager = SlotManager(slot_size)
         # This class is reused by MTP_EAGLE
         from ...llmapi.llm_args import EagleDecodingConfig
@@ -104,6 +113,7 @@ def __init__(self,
                 max_total_draft_tokens=self.max_total_draft_tokens,
                 eagle_choices=config.eagle_choices,
                 dynamic_tree_max_topK=config.dynamic_tree_max_topK,
+                num_seq_slots=self.num_seq_slots,
             )
 
     def prepare_resources(self, scheduled_batch: ScheduledRequests):
@@ -165,8 +175,14 @@ class Eagle3OneModelDynamicTreeResourceManager(BaseResourceManager):
     hidden_states: Optional[torch.Tensor] = None
     batch_indices_cuda: Optional[torch.Tensor] = None
 
-    def __init__(self, config: "EagleDecodingConfig", max_num_requests: int):
+    def __init__(self,
+                 config: "EagleDecodingConfig",
+                 max_num_requests: int,
+                 num_seq_slots: Optional[int] = None):
         self.max_num_requests = max_num_requests
+        # batch_indices_cuda is indexed by batch position, so it stays at
+        # max_batch_size; only the SpecTreeManager slot storage below is keyed by
+        # py_seq_slot and needs the executor's slot pool (nvbug-6627795).
         self.batch_indices_cuda = torch.empty(
             [max_num_requests],
             dtype=torch.int,
@@ -179,6 +195,7 @@ def __init__(self, config: "EagleDecodingConfig", max_num_requests: int):
             max_total_draft_tokens=config.tokens_per_gen_step - 1,
             eagle_choices=config.eagle_choices,
             dynamic_tree_max_topK=config.dynamic_tree_max_topK,
+            num_seq_slots=num_seq_slots,
         )
 
     def free_resources(self, request: LlmRequest):
diff --git a/tensorrt_llm/_torch/speculative/mtp.py b/tensorrt_llm/_torch/speculative/mtp.py
index 685143e2d975..f815a31fa181 100644
--- a/tensorrt_llm/_torch/speculative/mtp.py
+++ b/tensorrt_llm/_torch/speculative/mtp.py
@@ -42,15 +42,26 @@ def __init__(self,
                  dtype: torch.dtype,
                  hidden_size: int,
                  max_num_requests: int,
-                 sa_manager=None):
+                 sa_manager=None,
+                 num_seq_slots: Optional[int] = None):
         self.dtype = dtype
         self.num_draft_slots = config.max_draft_len
         self.hidden_size = hidden_size
         self.max_num_requests = max_num_requests
         self.use_relaxed_acceptance_for_thinking = config.use_relaxed_acceptance_for_thinking
+        # These slots are keyed by live-request identity, not by batch position:
+        # add_slot runs on a request's first context chunk and the slot is only
+        # returned by free_resources. So the pool must cover every request that
+        # can be resident at once, which is the SeqSlotManager pool size
+        # (``num_seq_slots``) rather than max_batch_size -- under the attention-DP
+        # overlap headroom the two differ by 2x, because a finished request holds
+        # its slot for one more iteration while its replacement is already
+        # admitted (nvbug-6627795). Sizing this at max_num_requests instead makes
+        # SlotManager.add_slot raise NoFreeSlotsError. Falls back to
+        # max_num_requests when the caller does not know the pool size.
         # Reserve one extra slot for the CUDA graph padding dummy request,
         # which is kept alive permanently and must not consume a real slot.
-        slot_pool_size = max_num_requests + 1
+        slot_pool_size = (num_seq_slots or max_num_requests) + 1
         self.slot_manager = SlotManager(slot_pool_size)
         # Optional SA manager for MTP+SA mode
         self.sa_manager = sa_manager
diff --git a/tensorrt_llm/_torch/speculative/mtp_dynamic_tree.py b/tensorrt_llm/_torch/speculative/mtp_dynamic_tree.py
index 86895f4c3558..021478a5fbec 100644
--- a/tensorrt_llm/_torch/speculative/mtp_dynamic_tree.py
+++ b/tensorrt_llm/_torch/speculative/mtp_dynamic_tree.py
@@ -1102,6 +1102,7 @@ def __init__(
         hidden_size: int,
         max_num_requests: int,
         sa_manager=None,
+        num_seq_slots: Optional[int] = None,
     ):
         from .spec_tree_manager import SpecTreeManager
 
@@ -1113,10 +1114,18 @@ def __init__(
             max_total_draft_tokens=config.tokens_per_gen_step - 1,
             eagle_choices=None,
             dynamic_tree_max_topK=config.dynamic_tree_max_topK,
+            num_seq_slots=num_seq_slots,
         )
         # MTP hidden-state slot pools (needed by MTPEagleWorker drafter inputs).
+        # num_seq_slots is forwarded because those pools are keyed by live-request
+        # identity; see MTPHiddenStatesManager.__init__.
         self._mtp_hidden_states_manager = MTPHiddenStatesManager(
-            config, dtype, hidden_size, max_num_requests, sa_manager=sa_manager
+            config,
+            dtype,
+            hidden_size,
+            max_num_requests,
+            sa_manager=sa_manager,
+            num_seq_slots=num_seq_slots,
         )
 
     # Expose the MTPHiddenStatesManager surface MTPSpecMetadata expects.
diff --git a/tensorrt_llm/_torch/speculative/spec_tree_manager.py b/tensorrt_llm/_torch/speculative/spec_tree_manager.py
index 545b5bc7bb06..d82c34cc7479 100644
--- a/tensorrt_llm/_torch/speculative/spec_tree_manager.py
+++ b/tensorrt_llm/_torch/speculative/spec_tree_manager.py
@@ -1,7 +1,7 @@
 import logging
 import math
 from itertools import accumulate
-from typing import List
+from typing import List, Optional
 
 import torch
 
@@ -231,10 +231,14 @@ class SpecTreeManager:
     retrieve_next_sibling: torch.Tensor = None
     slot_storage: 'DynamicTreeSlotStorage | None' = None
 
-    def __init__(self, max_num_requests: int, use_dynamic_tree: bool,
-                 max_total_draft_tokens: int, max_draft_len: int,
+    def __init__(self,
+                 max_num_requests: int,
+                 use_dynamic_tree: bool,
+                 max_total_draft_tokens: int,
+                 max_draft_len: int,
                  eagle_choices: List[List[int]] | None,
-                 dynamic_tree_max_topK: int):
+                 dynamic_tree_max_topK: int,
+                 num_seq_slots: Optional[int] = None):
 
         self.use_dynamic_tree = use_dynamic_tree
         self.max_total_draft_tokens = max_total_draft_tokens
@@ -251,6 +255,15 @@ def __init__(self, max_num_requests: int, use_dynamic_tree: bool,
             self._internal_buf_dim = max_total_draft_tokens + 1
         self.eagle_choices = eagle_choices
         self.num_trees = max_num_requests if use_dynamic_tree else 1
+        # ``num_trees`` sizes the per-forward *work* buffers, which are indexed by
+        # batch position and so correctly stay at max_batch_size -- the micro-batch
+        # scheduler caps every forward there. ``num_slots`` sizes DynamicTreeSlotStorage,
+        # which is indexed by ``py_seq_slot`` and must therefore span the executor's
+        # sequence-slot pool: the attention-DP overlap headroom makes that
+        # 2 * max_batch_size so a retiring request can keep its slot for one more
+        # iteration while its replacement is admitted (nvbug-6627795).
+        # None preserves the historical max_batch_size sizing.
+        self.num_slots = max(num_seq_slots or 0, max_num_requests)
         self.dynamic_tree_max_topK = dynamic_tree_max_topK
         self.cur_draft_layer_idx = 0
         self.top_k_list = []
@@ -334,7 +347,7 @@ def init_tree_info_for_dynamic_tree(self):
 
         mask_width = math.ceil(num_draft_with_root / 32)
         self.slot_storage = DynamicTreeSlotStorage(
-            num_slots=self.num_trees,
+            num_slots=self.num_slots,
             n_dt=num_draft_with_root,
             mask_width=mask_width,
             top_k=self.dynamic_tree_max_topK,
diff --git a/tensorrt_llm/_torch/speculative/suffix_automaton.py b/tensorrt_llm/_torch/speculative/suffix_automaton.py
index 4eba0781d46f..af413526a8b3 100644
--- a/tensorrt_llm/_torch/speculative/suffix_automaton.py
+++ b/tensorrt_llm/_torch/speculative/suffix_automaton.py
@@ -106,6 +106,7 @@ def __init__(
         config,
         max_num_requests: int,
         max_seq_len: int = 262144,
+        num_seq_slots: Optional[int] = None,
     ):
         if _sa_native is None:
             raise RuntimeError(
@@ -144,14 +145,27 @@ def __init__(
         self.max_seq_len = sa_config.max_seq_len
         self.enable_global_pool = sa_config.enable_global_pool
 
-        # Pool sizing: effective_pool_size returns max_num_requests when
-        # global pool is off, or max(64, max_num_requests) / explicit
-        # value when on. All slot-indexed sizing uses pool_size.
-        self.pool_size = sa_config.effective_pool_size
-        if self.pool_size < max_num_requests:
-            raise ValueError(
-                f"global_pool_size ({self.pool_size}) must be >= "
-                f"max_batch_size ({max_num_requests})"
+        # A slot is held for the whole lifetime of a request id, so the pool has to
+        # cover every request that can be simultaneously live -- that is the
+        # executor's sequence-slot pool, which the attention-DP overlap headroom
+        # raises to 2 * max_batch_size so a retiring request can keep its slot for
+        # one more iteration while its replacement is admitted (nvbug-6627795).
+        # None (no headroom) leaves this at max_batch_size, as before.
+        self._num_seq_slots = max(num_seq_slots or 0, max_num_requests)
+
+        # Pool sizing: effective_pool_size returns max_slots when global pool is
+        # off, or max(64, max_slots) / the explicit value when on. All slot-indexed
+        # sizing uses pool_size, so the live-slot count is a floor on it. An
+        # explicit global_pool_size below that floor is grown rather than rejected:
+        # the same config is legal without the headroom, so failing here would turn
+        # enabling attention DP into a startup error for a value that
+        # TorchLlmArgs.validate_speculative_config already accepted.
+        self.pool_size = max(sa_config.effective_pool_size, self._num_seq_slots)
+        if sa_config.global_pool_size is not None and self.pool_size > sa_config.global_pool_size:
+            logger.warning(
+                f"Growing the SA pool from the configured global_pool_size "
+                f"({sa_config.global_pool_size}) to {self.pool_size} to cover the "
+                f"executor's sequence slots."
             )
 
         # Calculate per-state size based on max_seq_len
@@ -159,7 +173,7 @@ def __init__(
 
         logger.info(
             f"SA pool: {self.pool_size} slots "
-            f"({self.pool_size - max_num_requests} retained capacity, "
+            f"({self.pool_size - self._num_seq_slots} retained capacity, "
             f"{self.pool_size * self.state_size / 1024 / 1024:.1f} MB total)"
         )
 
diff --git a/tensorrt_llm/_torch/speculative/utils.py b/tensorrt_llm/_torch/speculative/utils.py
index 849a045108b2..194cb4ea5fbd 100644
--- a/tensorrt_llm/_torch/speculative/utils.py
+++ b/tensorrt_llm/_torch/speculative/utils.py
@@ -339,12 +339,26 @@ def get_spec_metadata(spec_config,
                                     max_num_tokens,
                                     spec_resource_manager=spec_resource_manager,
                                     is_draft_model=is_draft_model,
-                                    max_seq_len=max_seq_len,
-                                    num_seq_slots=num_seq_slots)
+                                    max_seq_len=max_seq_len)
     # Set here rather than in each branch below: every one-model mode needs it and
     # the per-mode constructors are easy to miss one of.
     if metadata is not None:
         metadata.enable_penalty = getattr(spec_config, "enable_penalty", False)
+        # Same reasoning for the sequence-slot pool size, which sizes every
+        # slot-indexed buffer (draft_probs, full_draft_probs, penalty_state) and
+        # the dummy scratch row appended after them. It used to be forwarded by
+        # the MTP-eagle branch alone, so every other one-engine mode -- vanilla
+        # MTP, Eagle3 one-model, PARD, DFlash/DSpark, draft-target one-model --
+        # sized those buffers at max_num_requests while py_seq_slot ranged over
+        # the wider pool, indexing past the end of the allocation.
+        #
+        # Assigning after construction is in time: both consumers allocate
+        # lazily, from prepare() and from update_one_model_sampling_state, never
+        # from __post_init__. Leaving the field at its 0 default when the caller
+        # passes None keeps the established `num_seq_slots or max_num_requests`
+        # fallback in those two allocators.
+        if num_seq_slots is not None:
+            metadata.num_seq_slots = num_seq_slots
     return metadata
 
 
@@ -354,14 +368,11 @@ def _build_spec_metadata(spec_config,
                          max_num_tokens,
                          spec_resource_manager=None,
                          is_draft_model=False,
-                         max_seq_len=262144,
-                         num_seq_slots=None):
+                         max_seq_len=262144):
+    """Construct the per-mode metadata. The slot-pool size is applied by the
+    caller (``get_spec_metadata``) so no branch can forget it."""
     use_rejection_sampling = getattr(spec_config, "use_rejection_sampling",
                                      False)
-    # Slot-indexed buffers (draft_probs) must span the SeqSlotManager pool;
-    # DeepSeek-V4 overlap can exceed max_num_requests.
-    num_seq_slots = (num_seq_slots
-                     if num_seq_slots is not None else max_num_requests)
     vocab_size = getattr(model_config, "vocab_size", 0)
     # Draft-model vocab size, used to gate the d2t-expanded full_draft_probs
     # buffer allocation (see SpecMetadata.prepare_rejection_sampling_buffers).
@@ -385,7 +396,6 @@ def _build_spec_metadata(spec_config,
             use_rejection_sampling=use_rejection_sampling,
             advanced_sampling_mode=spec_config.advanced_sampling_mode,
             vocab_size=vocab_size,
-            num_seq_slots=num_seq_slots,
             draft_vocab_size=draft_vocab_size,
             spec_resource_manager=spec_resource_manager,
             use_dynamic_tree=getattr(spec_config, 'use_dynamic_tree', False),
@@ -562,6 +572,24 @@ def get_mtp_hidden_size(model_config) -> int:
     return hidden_size
 
 
+def seat_pool_or_none(model_engine) -> Optional[int]:
+    """The engine's sequence-slot pool size, or None to keep max_batch_size.
+
+    Pools keyed by live-request identity must follow the executor's
+    SeqSlotManager pool rather than max_batch_size: the overlap scheduler holds a
+    finished request's slot for one more iteration while its replacement is
+    admitted, so the transient demand exceeds max_batch_size (nvbug-6627795).
+    Buffers indexed by *batch position* deliberately keep max_batch_size -- the
+    micro-batch scheduler caps every forward at max_batch_size.
+
+    Gated on the same flag ``_set_up_spec_metadata`` reads, so every spec-decoding
+    pool agrees with the metadata about which number it is indexed by.
+    """
+    if not getattr(model_engine, "_enable_disagg_adp_overlap_headroom", False):
+        return None
+    return getattr(model_engine, "max_num_seq_slots", None)
+
+
 def get_spec_resource_manager(model_engine, draft_model_engine=None):
     spec_config = model_engine.spec_config
     if spec_config is None:
@@ -570,13 +598,16 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
     max_num_requests = model_engine.batch_size
     max_seq_len = model_engine.max_seq_len
     max_num_tokens = model_engine.max_num_tokens
+    num_seq_slots = seat_pool_or_none(model_engine)
     spec_dec_mode = spec_config.spec_dec_mode
     if spec_dec_mode.is_mtp_eagle_one_model():
         sa_manager = None
         sa_cfg = getattr(spec_config, 'sa_config', None)
         if sa_cfg is not None:
-            sa_manager = SuffixAutomatonManager(sa_cfg, max_num_requests,
-                                                max_seq_len)
+            sa_manager = SuffixAutomatonManager(sa_cfg,
+                                                max_num_requests,
+                                                max_seq_len,
+                                                num_seq_slots=num_seq_slots)
         # Dynamic tree combines SpecTreeManager with MTP hidden-state slots.
         if getattr(spec_config, 'use_dynamic_tree', False):
             return MTPEagleDynamicTreeResourceManager(
@@ -585,6 +616,7 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
                 get_mtp_hidden_size(model_config),
                 max_num_requests,
                 sa_manager=sa_manager,
+                num_seq_slots=num_seq_slots,
             )
         if spec_config.use_relaxed_acceptance_for_thinking or sa_manager is not None:
             # Unified resource manager: the unified worker reads
@@ -598,6 +630,7 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
                 max_seq_len,
                 max_num_tokens,
                 sa_manager=sa_manager,
+                num_seq_slots=num_seq_slots,
             )
         else:
             return None
@@ -605,25 +638,30 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
         sa_manager = None
         sa_cfg = getattr(spec_config, 'sa_config', None)
         if sa_cfg is not None:
-            sa_manager = SuffixAutomatonManager(sa_cfg, max_num_requests,
-                                                max_seq_len)
+            sa_manager = SuffixAutomatonManager(sa_cfg,
+                                                max_num_requests,
+                                                max_seq_len,
+                                                num_seq_slots=num_seq_slots)
         return MTPHiddenStatesManager(
             spec_config,
             model_config.torch_dtype,
             get_mtp_hidden_size(model_config),
             max_num_requests,
             sa_manager=sa_manager,
+            num_seq_slots=num_seq_slots,
         )
     if spec_dec_mode.is_eagle3_one_model() and _is_effective_dynamic_tree(
             spec_config):
-        return Eagle3OneModelDynamicTreeResourceManager(spec_config,
-                                                        max_num_requests)
+        return Eagle3OneModelDynamicTreeResourceManager(
+            spec_config, max_num_requests, num_seq_slots=num_seq_slots)
     if spec_dec_mode.is_eagle3_one_model():
         sa_manager = None
         sa_cfg = getattr(spec_config, 'sa_config', None)
         if sa_cfg is not None:
-            sa_manager = SuffixAutomatonManager(sa_cfg, max_num_requests,
-                                                max_seq_len)
+            sa_manager = SuffixAutomatonManager(sa_cfg,
+                                                max_num_requests,
+                                                max_seq_len,
+                                                num_seq_slots=num_seq_slots)
         return Eagle3ResourceManager(
             spec_config,
             model_config.torch_dtype,
@@ -632,6 +670,7 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
             max_seq_len,
             max_num_tokens,
             sa_manager=sa_manager,
+            num_seq_slots=num_seq_slots,
         )
     if spec_dec_mode.is_eagle3() or spec_dec_mode.is_mtp_eagle():
         assert draft_model_engine is not None, "Draft model engine is required for Eagle3 and MTP Eagle two model flow."
@@ -642,6 +681,7 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
             max_num_requests,
             max_seq_len,
             max_num_tokens,
+            num_seq_slots=num_seq_slots,
         )
     if spec_dec_mode.is_save_hidden_states():
         return SaveHiddenStatesResourceManager(
@@ -654,13 +694,18 @@ def get_spec_resource_manager(model_engine, draft_model_engine=None):
     if spec_dec_mode.is_parallel_draft():
         sa_cfg = getattr(spec_config, 'sa_config', None)
         if sa_cfg is not None:
-            return SuffixAutomatonManager(sa_cfg, max_num_requests, max_seq_len)
+            return SuffixAutomatonManager(sa_cfg,
+                                          max_num_requests,
+                                          max_seq_len,
+                                          num_seq_slots=num_seq_slots)
         return None
     if spec_dec_mode.is_ngram():
         return NGramPoolManager(spec_config, max_num_requests)
     if spec_dec_mode.is_sa():
-        return SuffixAutomatonManager(spec_config, max_num_requests,
-                                      max_seq_len)
+        return SuffixAutomatonManager(spec_config,
+                                      max_num_requests,
+                                      max_seq_len,
+                                      num_seq_slots=num_seq_slots)
     if spec_dec_mode.is_user_provided():
         return spec_config.resource_manager
     return None
@@ -721,6 +766,14 @@ def get_spec_drafter(model_engine,
         return spec_config.drafter
 
     max_num_requests = model_engine.batch_size
+    # The draft loop runs its own slot pool, but the indices it hands out address
+    # buffers sized by the *target* engine's seat pool (the shared sampler and
+    # spec_resource_manager above), so it must be sized from the same number. The
+    # previous draft batch's slots are also released by
+    # cleanup_previous_draft_resources a full iteration later, so a pool of
+    # max_batch_size raises NoFreeSlotsError precisely when the overlap headroom
+    # is doing its job.
+    draft_slots = seat_pool_or_none(model_engine) or max_num_requests
     if spec_config.spec_dec_mode.is_draft_target(
     ) or spec_config.spec_dec_mode.is_eagle3(
     ) or spec_config.spec_dec_mode.is_mtp_eagle():
@@ -728,7 +781,7 @@ def get_spec_drafter(model_engine,
                             draft_model_engine,
                             spec_config.max_draft_len,
                             spec_config.tokens_per_gen_step - 1,
-                            SeqSlotManager(max_num_requests),
+                            SeqSlotManager(draft_slots),
                             sampler,
                             spec_resource_manager=spec_resource_manager,
                             guided_decoder=guided_decoder)
diff --git a/tests/unittest/_torch/disaggregation/test_benchmark_disagg.py b/tests/unittest/_torch/disaggregation/test_benchmark_disagg.py
index 79eabcd99193..dc0be2ec8a1a 100644
--- a/tests/unittest/_torch/disaggregation/test_benchmark_disagg.py
+++ b/tests/unittest/_torch/disaggregation/test_benchmark_disagg.py
@@ -803,6 +803,11 @@ def __init__(
         self.resource_manager = Mock()
         self.resource_manager.get_resource_manager.return_value = None
 
+        # The pad path reads the router's gate for "which requests count as
+        # routable load"; non-PP attention DP excludes the retiring ones
+        # (nvbug-6627795).
+        self.adp_router = Mock(exclude_retiring_requests=True)
+
     from tensorrt_llm._torch.pyexecutor.py_executor import PyExecutor, _ADPForwardIntent
 
     _pad_attention_dp_dummy_request = PyExecutor._pad_attention_dp_dummy_request
diff --git a/tests/unittest/_torch/executor/kv_cache/test_kv_cache_estimation.py b/tests/unittest/_torch/executor/kv_cache/test_kv_cache_estimation.py
index 6bad069afaa4..efe1732f9c13 100644
--- a/tests/unittest/_torch/executor/kv_cache/test_kv_cache_estimation.py
+++ b/tests/unittest/_torch/executor/kv_cache/test_kv_cache_estimation.py
@@ -1158,6 +1158,10 @@ def test_separate_one_model_draft_normalizes_target_pool_ratio() -> None:
     creator._is_disagg = False
     creator._mapping = Mock()
     creator._speculative_config = Mock()
+    # Every manager is now told whether the overlap scheduler is off, so this
+    # hand-built creator has to carry the attribute the real one sets in
+    # __init__. The value does not move this test's assertions.
+    creator._disable_overlap_scheduler = False
 
     effective_draft_config = Mock()
     effective_draft_config.pretrained_config.torch_dtype = "bfloat16"
diff --git a/tests/unittest/_torch/executor/kv_cache/test_kv_cache_manager_v2.py b/tests/unittest/_torch/executor/kv_cache/test_kv_cache_manager_v2.py
index 1212a39e8dc3..5c0ff295f729 100644
--- a/tests/unittest/_torch/executor/kv_cache/test_kv_cache_manager_v2.py
+++ b/tests/unittest/_torch/executor/kv_cache/test_kv_cache_manager_v2.py
@@ -1542,3 +1542,136 @@ def test_disagg_role_mapper_kinds_default_to_indexed():
         Role.ALL: MapperKind.INDEXED,
         Role.INDEX_KEY: MapperKind.REPLICATED,
     }
+
+
+def _index_mapper_capacity_for(
+    *,
+    max_batch_size: int,
+    pp_size: int = 1,
+    is_disagg: bool = False,
+    num_reserved_index_slots: int = 1,
+    enable_attention_dp: bool = False,
+    disable_overlap_scheduler: bool = False,
+) -> tuple[int, int]:
+    """Construct a manager and return (IndexMapper capacity, page-table capacity).
+
+    The two must agree: ``host_kv_cache_block_offsets`` is indexed by the index the
+    mapper hands out, so a page table sized below the mapper's capacity would be an
+    out-of-bounds write.
+    """
+    module = "tensorrt_llm._torch.pyexecutor.kv_cache.kv_cache_manager_v2"
+    fake_impl = Mock()
+    fake_impl.layer_grouping = [[0]]
+    fake_impl.pool_group_descs = []
+    fake_impl.get_layer_group_id.side_effect = lambda _: 0
+
+    def build_base_config(
+        self: KVCacheManagerV2,
+        config: KvCacheConfig,
+        *,
+        tokens_per_block: int,
+        cache_tiers: list[object],
+    ) -> _FakeManagerConfig:
+        del self, config, tokens_per_block
+        return _FakeManagerConfig(cache_tiers=cache_tiers)
+
+    with (
+        patch(f"{module}.IndexMapper") as index_mapper_cls,
+        patch(f"{module}.KVCacheManagerPy", Mock(return_value=fake_impl)),
+        patch.object(KVCacheManagerV2, "_build_base_config", build_base_config),
+        patch.object(KVCacheManagerV2, "_build_cache_config", lambda self, config: config),
+        patch.object(KVCacheManagerV2, "get_num_available_tokens", return_value=MAX_SEQ_LEN),
+        patch.object(KVCacheManagerV2, "_prepare_page_table_tensor") as page_table,
+        patch.object(KVCacheManagerV2, "_log_kv_cache_pool_lifecycle_mapping"),
+    ):
+        KVCacheManagerV2(
+            # A quota must be set or __init__ asserts before it sizes anything
+            # ("Quota not set. Check kv_cache_config.max_tokens or
+            # kv_cache_config.max_gpu_total_bytes"). The value is irrelevant to the
+            # index-mapper arithmetic, which reads only max_batch_size, pp_size,
+            # enable_attention_dp, is_disagg, disable_overlap_scheduler and
+            # num_reserved_index_slots.
+            KvCacheConfig(max_gpu_total_bytes=16 << 20),
+            CacheType.SELFKONLY,
+            num_layers=1,
+            num_kv_heads=1,
+            head_dim=1,
+            tokens_per_block=TOKENS_PER_BLOCK,
+            max_seq_len=MAX_SEQ_LEN,
+            max_batch_size=max_batch_size,
+            mapping=Mapping(
+                world_size=pp_size,
+                rank=0,
+                tp_size=1,
+                pp_size=pp_size,
+                enable_attention_dp=enable_attention_dp,
+            ),
+            dtype=DataType.HALF,
+            vocab_size=16,
+            execution_stream=Mock(),
+            is_disagg=is_disagg,
+            num_reserved_index_slots=num_reserved_index_slots,
+            disable_overlap_scheduler=disable_overlap_scheduler,
+        )
+    index_mapper_cls.assert_called_once()
+    page_table.assert_called_once()
+    return (
+        index_mapper_cls.call_args.args[0],
+        page_table.call_args.args[0],
+    )
+
+
+# (max_batch_size, pp_size, adp, is_disagg, disable_overlap, reserved, expected)
+#
+# capacity == max_batch_size * pp_size * (2 if is_disagg or (adp and not
+# disable_overlap and not pp) else 1) + reserved. Rows 1-2 are the nvbug 6627795
+# case: under attention DP the overlap scheduler defers the retiring batch's
+# teardown past the point where its replacement is admitted, so both cohorts hold
+# index slots at once and a mapper sized at B+1 silently defers requests one at a
+# time.
+_INDEX_MAPPER_CAPACITY_CASES = [
+    # ADP + overlap, no PP: both cohorts are resident, so the mapper needs 2B.
+    pytest.param(2, 1, True, False, False, 1, 5, id="adp_overlap"),
+    pytest.param(8, 1, True, False, False, 1, 17, id="adp_overlap_b8"),
+    # Either half of the conjunction missing leaves the pre-fix allocation.
+    pytest.param(2, 1, True, False, True, 1, 3, id="adp_no_overlap"),
+    pytest.param(2, 1, False, False, False, 1, 3, id="overlap_no_adp"),
+    # Disagg already carried its own 2x; the two coefficients cover the same
+    # extra cohort, so they do not compound.
+    pytest.param(2, 1, True, True, False, 1, 5, id="disagg_does_not_compound"),
+    pytest.param(2, 1, False, True, True, 1, 5, id="disagg_only"),
+    # Pipeline parallelism is out of scope: max_batch_size * pp_size already
+    # covers the in-flight micro-batches, so the ADP coefficient stays off.
+    pytest.param(2, 4, True, False, False, 1, 9, id="pp4_adp_overlap"),
+    pytest.param(2, 4, False, False, True, 1, 9, id="pp4_plain"),
+    # ... while the pre-existing disagg 2x under PP is left exactly as it was.
+    pytest.param(2, 4, False, True, True, 1, 17, id="pp4_disagg_unchanged"),
+    # Reserved slots are still added on top of the widened pool.
+    pytest.param(2, 1, True, False, False, 5, 9, id="reserved_slots_still_added"),
+]
+
+
+@pytest.mark.cpu_only
+@pytest.mark.parametrize(
+    "max_batch_size,pp_size,adp,is_disagg,disable_overlap,reserved,expected",
+    _INDEX_MAPPER_CAPACITY_CASES,
+)
+def test_index_mapper_capacity_covers_the_overlapping_cohorts(
+    max_batch_size: int,
+    pp_size: int,
+    adp: bool,
+    is_disagg: bool,
+    disable_overlap: bool,
+    reserved: int,
+    expected: int,
+) -> None:
+    capacity, page_table_capacity = _index_mapper_capacity_for(
+        max_batch_size=max_batch_size,
+        pp_size=pp_size,
+        is_disagg=is_disagg,
+        num_reserved_index_slots=reserved,
+        enable_attention_dp=adp,
+        disable_overlap_scheduler=disable_overlap,
+    )
+    assert capacity == expected
+    assert page_table_capacity == expected
diff --git a/tests/unittest/_torch/executor/test_adp_router.py b/tests/unittest/_torch/executor/test_adp_router.py
index 250b214c86e1..38b9bae8dd15 100644
--- a/tests/unittest/_torch/executor/test_adp_router.py
+++ b/tests/unittest/_torch/executor/test_adp_router.py
@@ -7,11 +7,13 @@
 - Strict/relaxed attention-DP request routing while respecting rank capacity
 """
 
+import inspect
 from unittest.mock import MagicMock, Mock
 
 import pytest
 
 from tensorrt_llm._torch.pyexecutor.executor_request_queue import RequestQueueItem
+from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequestState
 from tensorrt_llm._torch.pyexecutor.request_utils import get_from_waiting_queue
 from tensorrt_llm._torch.pyexecutor.scheduler import FCFSWaitingQueue
 from tensorrt_llm._torch.pyexecutor.scheduler.adp_router import (
@@ -22,6 +24,9 @@
     RankIterStatsPayload,
     RankState,
     _num_input_tokens,
+    build_active_requests_for_overlap,
+    count_retiring_requests,
+    is_retiring_request,
 )
 from tensorrt_llm.conversation_params import ConversationParams
 from tensorrt_llm.scheduling_params import SchedulingParams
@@ -40,12 +45,16 @@ def num_input_tokens(self):
         return len(self.input_token_ids)
 
 
-def _mock_dist(tp_rank=0, tp_size=1, has_cp_helix=False):
+def _mock_dist(tp_rank=0, tp_size=1, has_cp_helix=False, has_pp=False):
     """Create a mock Distributed object for testing."""
     dist = MagicMock()
     dist.tp_rank = tp_rank
     dist.tp_size = tp_size
     dist.has_cp_helix = has_cp_helix
+    # ADPRouter reads this to decide whether to route on the overlap-corrected
+    # active list; a bare MagicMock would make has_pp() truthy and silently
+    # disable the correction in every test.
+    dist.mapping.has_pp.return_value = has_pp
     return dist
 
 
@@ -130,6 +139,74 @@ def all_ranks_num_active_tokens():
     return [10, 5, 15, 8]
 
 
+def _retiring_request(prompt_len=100):
+    """Active request that has produced its final token (state 14)."""
+    return Mock(
+        py_orig_prompt_len=prompt_len,
+        cached_tokens=0,
+        state=LlmRequestState.GENERATION_TO_COMPLETE,
+    )
+
+
+class TestBuildActiveRequestsForOverlap:
+    # Retiring requests linger in active_requests for one extra iteration when
+    # the overlap scheduler is on. They must not be charged against ADP
+    # admission capacity, because no scheduler can ever forward them again
+    # (nvbug-6627795). Every other lingering state still owns a seat and KV, so
+    # it must keep counting.
+    def test_empty(self):
+        assert build_active_requests_for_overlap([]) == []
+        assert count_retiring_requests([]) == 0
+
+    def test_drops_generation_to_complete(self):
+        reqs = [_retiring_request(), _retiring_request(), _retiring_request()]
+        assert build_active_requests_for_overlap(reqs) == []
+        assert count_retiring_requests(reqs) == 3
+
+    @pytest.mark.parametrize(
+        "state",
+        [
+            LlmRequestState.CONTEXT_INIT,
+            LlmRequestState.GENERATION_IN_PROGRESS,
+            LlmRequestState.DISAGG_CONTEXT_WAIT_SCHEDULER,
+            LlmRequestState.DISAGG_GENERATION_INIT,
+            LlmRequestState.DISAGG_GENERATION_TRANS_IN_PROGRESS,
+            LlmRequestState.DISAGG_CONTEXT_TRANS_IN_PROGRESS,
+            LlmRequestState.DISAGG_TRANS_ERROR,
+        ],
+    )
+    def test_other_states_still_count_as_load(self, state):
+        req = Mock(state=state)
+        assert is_retiring_request(req) is False
+        assert build_active_requests_for_overlap([req]) == [req]
+        assert count_retiring_requests([req]) == 0
+
+    def test_mixed_preserves_order_of_survivors(self):
+        keep_a = Mock(state=LlmRequestState.GENERATION_IN_PROGRESS)
+        keep_b = Mock(state=LlmRequestState.DISAGG_GENERATION_INIT)
+        reqs = [keep_a, _retiring_request(), keep_b, _retiring_request()]
+        assert build_active_requests_for_overlap(reqs) == [keep_a, keep_b]
+        assert count_retiring_requests(reqs) == 2
+
+    def test_returns_a_new_list(self):
+        # The filtered list is the ROUTER's view only; mutating it must never
+        # disturb PyExecutor.active_requests, whose teardown ordering is what
+        # created the bug in the first place.
+        reqs = [Mock(state=LlmRequestState.GENERATION_IN_PROGRESS)]
+        filtered = build_active_requests_for_overlap(reqs)
+        assert filtered is not reqs
+        filtered.clear()
+        assert len(reqs) == 1
+
+    def test_bare_mock_is_not_retiring(self):
+        # Router tests build requests with bare Mocks that never set `state`.
+        # Identity comparison against the enum keeps those routable; a truthy
+        # bound-property check would drop every one of them.
+        req = Mock(py_orig_prompt_len=10)
+        assert build_active_requests_for_overlap([req]) == [req]
+        assert count_retiring_requests([req]) == 0
+
+
 class TestRankState:
     # RankState is the wire payload shared across attention-DP ranks. Keep its
     # serialization stable because iter-stats now ride on the same allgather.
@@ -141,7 +218,7 @@ def test_creation(self):
 
     def test_serialize(self):
         state = RankState(rank=0, num_active_requests=5, num_active_tokens=100)
-        assert state.serialize() == [0, 5, 100, 0, -1, 0, 0, 0, 0, 0, 0, 0]
+        assert state.serialize() == [0, 5, 100, 0, 0, -1, 0, 0, 0, 0, 0, 0, 0]
 
     def test_deserialize(self):
         state = RankState.deserialize(data=[2, 3, 50])
@@ -154,10 +231,22 @@ def test_roundtrip(self):
         restored = RankState.deserialize(data=original.serialize())
         assert original == restored
 
+    def test_roundtrip_with_retiring_requests(self):
+        original = RankState(
+            rank=1,
+            num_active_requests=10,
+            num_active_tokens=200,
+            num_retiring_requests=3,
+        )
+        restored = RankState.deserialize(data=original.serialize())
+        assert original == restored
+        assert restored.num_retiring_requests == 3
+
     def test_defaults(self):
         state = RankState(rank=0)
         assert state.num_active_requests == 0
         assert state.num_active_tokens == 0
+        assert state.num_retiring_requests == 0
         assert state.iter_stats.has_iter_stats == 0
         assert state.iter_stats.iter_stats_iter == -1
 
@@ -278,6 +367,149 @@ def test_create_rank_state_default(self):
         assert state.num_active_requests == 2
         assert state.num_active_tokens == 300
 
+    def test_create_rank_state_does_not_filter_retiring_itself(self):
+        # create_rank_state stays overlap-agnostic: it reports what it is given.
+        # The correction lives in gather_all_rank_states, so a router author
+        # cannot forget it.
+        dist = _mock_dist(tp_rank=0, has_cp_helix=False)
+        router = DefaultADPRouter(dist=dist)
+        active = [
+            Mock(py_orig_prompt_len=100, state=LlmRequestState.GENERATION_IN_PROGRESS),
+            _retiring_request(prompt_len=200),
+        ]
+        state = router.create_rank_state(active_requests=active, new_requests=[])
+        assert state.num_active_requests == 2
+        assert state.num_active_tokens == 300
+
+    def test_gather_all_rank_states_excludes_retiring(self):
+        # Two of three requests are retiring, so only one is routable load --
+        # and the tokens of the retiring pair go with them, because the filtered
+        # list is what create_rank_state sums.
+        dist = _mock_dist(tp_rank=0, has_cp_helix=False)
+        router = DefaultADPRouter(dist=dist)
+        active = [
+            Mock(py_orig_prompt_len=100, state=LlmRequestState.GENERATION_IN_PROGRESS),
+            _retiring_request(prompt_len=200),
+            _retiring_request(prompt_len=300),
+        ]
+        dist.tp_allgather.side_effect = lambda payload: [payload]
+
+        states = router.gather_all_rank_states(active_requests=active)
+
+        assert len(states) == 1
+        assert states[0].num_active_requests == 1
+        assert states[0].num_retiring_requests == 2
+        assert states[0].num_active_tokens == 100
+        # The executor's own list is untouched -- only the router's view narrows.
+        assert len(active) == 3
+
+    def test_gather_all_rank_states_reports_zero_when_all_retiring(self):
+        # Nothing routable, but the rank is NOT idle: the retiring requests are
+        # still resident. num_retiring_requests carries that fact to every peer
+        # so the idle-fetch wait stays collective (nvbug-6627795).
+        dist = _mock_dist(tp_rank=0, has_cp_helix=False)
+        router = DefaultADPRouter(dist=dist)
+        active = [_retiring_request(), _retiring_request()]
+        dist.tp_allgather.side_effect = lambda payload: [payload]
+
+        states = router.gather_all_rank_states(active_requests=active)
+
+        assert states[0].num_active_requests == 0
+        assert states[0].num_active_tokens == 0
+        assert states[0].num_retiring_requests == 2
+
+    def test_gather_all_rank_states_keeps_retiring_without_headroom(self):
+        # Without seat headroom the correction is off, whatever the topology.
+        # The correction lets a rank hold more requests than it is charged for,
+        # so it is only sound when the sequence-slot pool was sized with the
+        # extra generation of seats -- e.g. hybrid/SSM architectures are excluded
+        # from the headroom because their state-slot pool is sized from
+        # max_batch_size alone, and the router must follow that exclusion rather
+        # than re-derive its own predicate.
+        dist = _mock_dist(tp_rank=0, has_cp_helix=False, has_pp=True)
+        router = DefaultADPRouter(dist=dist, has_seq_slot_headroom=False)
+        assert router.exclude_retiring_requests is False
+        active = [
+            Mock(py_orig_prompt_len=100, state=LlmRequestState.GENERATION_IN_PROGRESS),
+            _retiring_request(prompt_len=200),
+            _retiring_request(prompt_len=300),
+        ]
+        dist.tp_allgather.side_effect = lambda payload: [payload]
+
+        states = router.gather_all_rank_states(active_requests=active)
+
+        # Same inputs as test_gather_all_rank_states_excludes_retiring, which
+        # asserts 1 / 2 / 100 -- here nothing is filtered.
+        assert states[0].num_active_requests == 3
+        assert states[0].num_retiring_requests == 0
+        assert states[0].num_active_tokens == 600
+
+    def test_exclude_retiring_requests_follows_the_seat_pool_headroom(self):
+        # The flag tracks the *engine's* headroom flag and nothing else, so the
+        # correction can only ever spend seats that were actually allocated. It
+        # used to re-derive the predicate as `not dist.mapping.has_pp()`, which
+        # is a second copy of a fact the sizing gate already owns.
+        for has_pp in (False, True):
+            dist = _mock_dist(has_pp=has_pp)
+            assert (
+                DefaultADPRouter(dist=dist, has_seq_slot_headroom=True).exclude_retiring_requests
+                is True
+            )
+            assert (
+                DefaultADPRouter(dist=dist, has_seq_slot_headroom=False).exclude_retiring_requests
+                is False
+            )
+
+    def test_router_factory_requires_the_headroom_flag(self):
+        # Required rather than defaulted: a caller that silently got the
+        # aggressive behaviour would be re-introducing the skew this parameter
+        # exists to remove. There is exactly one production call site
+        # (PyExecutor.__init__), which passes the engine's flag.
+        params = inspect.signature(ADPRouter.create).parameters
+        assert params["has_seq_slot_headroom"].default is inspect.Parameter.empty
+
+    @pytest.mark.parametrize("has_seq_slot_headroom", [True, False])
+    def test_router_factory_propagates_the_headroom_flag(self, has_seq_slot_headroom):
+        # Every branch of the factory, not just the default one: the KV-cache-
+        # aware and conversation-affinity routers run the same admission
+        # correction and need the same gate.
+        dist = _mock_dist(tp_size=2)
+        mgr = Mock(enable_block_reuse=True)
+        configs = [
+            None,
+            Mock(
+                kv_cache_routing_conversation_affinity=False,
+                enable_kv_cache_aware_routing=True,
+                kv_cache_routing_load_balance_weight=1.0,
+                kv_cache_routing_match_rate_threshold=0.1,
+                kv_cache_routing_fair_share_multiplier=2.0,
+                kv_cache_routing_cold_start_warmup=False,
+                kv_cache_routing_account_for_in_transfer=False,
+            ),
+            Mock(
+                kv_cache_routing_conversation_affinity=True,
+                kv_cache_routing_max_sessions=1 << 16,
+                kv_cache_routing_fair_share_multiplier=2.0,
+                kv_cache_routing_new_conv_placement="round_robin",
+            ),
+        ]
+        built = set()
+        for attention_dp_config in configs:
+            router = ADPRouter.create(
+                dist=dist,
+                has_seq_slot_headroom=has_seq_slot_headroom,
+                kv_cache_manager=mgr,
+                attention_dp_config=attention_dp_config,
+            )
+            built.add(type(router).__name__)
+            assert router.exclude_retiring_requests is has_seq_slot_headroom
+        # Anti-vacuity: the three configs must actually reach three branches.
+        assert built == {
+            "DefaultADPRouter",
+            "KVCacheAwareADPRouter",
+            "ConversationAwareADPRouter",
+        }
+
     def test_create_rank_state_cp_helix(self):
         dist = _mock_dist(tp_rank=1, has_cp_helix=True)
         router = DefaultADPRouter(dist=dist)
@@ -1254,11 +1486,36 @@ def test_create_rank_state(self):
         assert state.num_active_requests == 2
         assert state.num_active_tokens == 150
 
+    def test_gather_all_rank_states_excludes_retiring(self):
+        # The filter sits in the shared ADPRouter.gather_all_rank_states, so it
+        # applies to this router without a line of router-specific code.
+        dist = _mock_dist(tp_rank=2)
+        router = ConversationAwareADPRouter(dist=dist)
+        active = [
+            Mock(py_orig_prompt_len=100, state=LlmRequestState.GENERATION_IN_PROGRESS),
+            _retiring_request(prompt_len=50),
+        ]
+        dist.tp_allgather.side_effect = lambda payload: [payload]
+
+        states = router.gather_all_rank_states(active_requests=active)
+
+        assert states[0].num_active_requests == 1
+        assert states[0].num_retiring_requests == 1
+        assert states[0].num_active_tokens == 100
+
     def test_factory_selects_conversation_router(self):
+        # has_seq_slot_headroom is required rather than defaulted (see
+        # test_router_factory_requires_the_headroom_flag), so every call site states
+        # it -- including the ones that are only about which class gets built.
         cfg = MagicMock()
         cfg.kv_cache_routing_conversation_affinity = True
         cfg.kv_cache_routing_max_sessions = 8
-        router = ADPRouter.create(dist=_mock_dist(), kv_cache_manager=None, attention_dp_config=cfg)
+        router = ADPRouter.create(
+            dist=_mock_dist(),
+            has_seq_slot_headroom=True,
+            kv_cache_manager=None,
+            attention_dp_config=cfg,
+        )
         assert isinstance(router, ConversationAwareADPRouter)
         assert router._max_sessions == 8
         # A mocked (non-string) placement value must fall back to round_robin.
@@ -1313,7 +1570,12 @@ def test_new_conv_placement_config(self):
         cfg.kv_cache_routing_conversation_affinity = True
         cfg.kv_cache_routing_max_sessions = 8
         cfg.kv_cache_routing_new_conv_placement = "least_queued"
-        router = ADPRouter.create(dist=_mock_dist(), kv_cache_manager=None, attention_dp_config=cfg)
+        router = ADPRouter.create(
+            dist=_mock_dist(),
+            has_seq_slot_headroom=True,
+            kv_cache_manager=None,
+            attention_dp_config=cfg,
+        )
         assert router._new_conv_placement == "least_queued"
         bad = ConversationAwareADPRouter(dist=_mock_dist(tp_size=4), new_conv_placement="banana")
         assert bad._new_conv_placement == "round_robin"
@@ -1322,7 +1584,12 @@ def test_factory_default_when_disabled(self):
         cfg = MagicMock()
         cfg.kv_cache_routing_conversation_affinity = False
         cfg.enable_kv_cache_aware_routing = False
-        router = ADPRouter.create(dist=_mock_dist(), kv_cache_manager=None, attention_dp_config=cfg)
+        router = ADPRouter.create(
+            dist=_mock_dist(),
+            has_seq_slot_headroom=True,
+            kv_cache_manager=None,
+            attention_dp_config=cfg,
+        )
         assert isinstance(router, DefaultADPRouter)
 
     def test_returned_expected_covers_every_rank(self):
diff --git a/tests/unittest/_torch/executor/test_kvcache_aware_router.py b/tests/unittest/_torch/executor/test_kvcache_aware_router.py
index 1f173ed0da83..43bdcf72ec85 100644
--- a/tests/unittest/_torch/executor/test_kvcache_aware_router.py
+++ b/tests/unittest/_torch/executor/test_kvcache_aware_router.py
@@ -21,6 +21,7 @@
 
 import pytest
 
+from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequestState
 from tensorrt_llm._torch.pyexecutor.scheduler.adp_router import (
     ADPRouter,
     KVCacheAwareADPRouter,
@@ -33,7 +34,7 @@
 # ---- Helpers ----
 
 
-def _mock_dist(tp_rank=0, tp_size=1, has_cp_helix=False):
+def _mock_dist(tp_rank=0, tp_size=1, has_cp_helix=False, has_pp=False):
     """Create a mock Distributed object for testing."""
     dist = MagicMock()
     dist.tp_rank = tp_rank
@@ -41,6 +42,10 @@ def _mock_dist(tp_rank=0, tp_size=1, has_cp_helix=False):
     # ADP scheduling assumes ``enable_attention_dp=True``, so ``dp_size``
     # mirrors ``tp_size`` (see ``Mapping.dp_size``).
     dist.mapping.dp_size = tp_size
+    # ADPRouter reads this to decide whether to route on the overlap-corrected
+    # active list; a bare MagicMock would make has_pp() truthy and silently
+    # disable the correction in every test.
+    dist.mapping.has_pp.return_value = has_pp
     dist.has_cp_helix = has_cp_helix
     return dist
 
@@ -116,6 +121,63 @@ def test_create_rank_state(self):
         assert state.num_active_requests == 2
         assert state.num_active_tokens == 300
 
+    def test_gather_all_rank_states_excludes_retiring(self):
+        dist = _mock_dist(tp_rank=0)
+        mgr = _mock_kv_cache_manager()
+        router = KVCacheAwareADPRouter(dist=dist, kv_cache_manager=mgr)
+
+        req1 = Mock(
+            py_orig_prompt_len=100,
+            cached_tokens=0,
+            state=LlmRequestState.GENERATION_IN_PROGRESS,
+        )
+        req2 = Mock(
+            py_orig_prompt_len=200,
+            cached_tokens=0,
+            state=LlmRequestState.GENERATION_TO_COMPLETE,
+        )
+        dist.tp_allgather.side_effect = lambda payload: [payload]
+
+        states = router.gather_all_rank_states([req1, req2])
+
+        assert states[0].num_active_requests == 1
+        assert states[0].num_retiring_requests == 1
+        assert states[0].num_active_tokens == 100
+
+    def test_gather_all_rank_states_retiring_and_in_transfer(self):
+        # In-transfer requests have already left active_requests, so this router
+        # adds them back as load; retiring requests are still in it and are
+        # filtered out. The two corrections are independent and must compose.
+        dist = _mock_dist(tp_rank=0)
+        mgr = _mock_kv_cache_manager()
+        transfer_mgr = MagicMock()
+        in_transfer_req = Mock(py_orig_prompt_len=70, cached_tokens=0)
+        transfer_mgr.requests_in_transfer.return_value = {1: in_transfer_req}
+        router = KVCacheAwareADPRouter(
+            dist=dist,
+            kv_cache_manager=mgr,
+            async_transfer_manager=transfer_mgr,
+            account_for_in_transfer=True,
+        )
+
+        req1 = Mock(
+            py_orig_prompt_len=100,
+            cached_tokens=0,
+            state=LlmRequestState.GENERATION_IN_PROGRESS,
+        )
+        req2 = Mock(
+            py_orig_prompt_len=200,
+            cached_tokens=0,
+            state=LlmRequestState.GENERATION_TO_COMPLETE,
+        )
+        dist.tp_allgather.side_effect = lambda payload: [payload]
+
+        states = router.gather_all_rank_states([req1, req2])
+
+        assert states[0].num_active_requests == 2  # 1 routable + 1 in transfer
+        assert states[0].num_retiring_requests == 1
+        assert states[0].num_active_tokens == 170
+
     def test_create_rank_state_cp_helix(self):
         dist = _mock_dist(tp_rank=1, has_cp_helix=True)
         mgr = _mock_kv_cache_manager()
diff --git a/tests/unittest/_torch/executor/test_py_executor.py b/tests/unittest/_torch/executor/test_py_executor.py
index 8a659ecd7483..865e196ca8dc 100644
--- a/tests/unittest/_torch/executor/test_py_executor.py
+++ b/tests/unittest/_torch/executor/test_py_executor.py
@@ -1807,6 +1807,7 @@ def __init__(
         enable_scheduler_aware_adp_dummy=None,
         enable_non_overlap_adp_forward_intent=None,
         peer_forward_intent=_ADPForwardIntent.GENERATION,
+        exclude_retiring_requests=True,
     ):
         self.enable_attention_dp = enable_attention_dp
         self.kv_cache_transceiver = kv_cache_transceiver
@@ -1839,6 +1840,12 @@ def __init__(
         self.dist.tp_size = 1
         self.dist.tp_allgather.side_effect = lambda value: [value]
         self.dist.tp_allreduce.side_effect = lambda value, op: max(value, int(peer_forward_intent))
+        # The pad path reads the router's gate rather than re-deriving it, so
+        # the two views of "which requests are routable load" cannot drift.
+        # Default True models a non-PP attention-DP executor; under pipeline
+        # parallelism the router leaves retiring requests in the load vector
+        # and the pad path must not subtract them (nvbug-6627795).
+        self.adp_router = Mock(exclude_retiring_requests=exclude_retiring_requests)
 
         self.scheduler = Mock()
         self.scheduler.scheduling_state_range = (
@@ -2142,6 +2149,43 @@ def test_decoder_context_waiting_for_encoder_output_is_not_counted():
     assert len(stub.active_requests) == 2
 
 
+def test_pad_does_not_warn_when_surplus_is_only_retiring_requests():
+    # The router now excludes retiring requests from the per-rank loads that
+    # floor `expected` (nvbug-6627795), so `expected` can legitimately sit
+    # below len(active_requests). Measuring the surplus against the raw len()
+    # would log a warning on every iteration of a hot loop.
+    stub = _StubADPExecutor()
+    stub.active_requests = [
+        _make_adp_request(_STATE_GENERATION_IN_PROGRESS, request_id=1),
+        _make_adp_request(_STATE_GENERATION_TO_COMPLETE, request_id=2),
+        _make_adp_request(_STATE_GENERATION_TO_COMPLETE, request_id=3),
+    ]
+    # What the router would have reported: 3 resident, 1 routable.
+    stub.expected_num_active_requests = 1
+
+    with patch("tensorrt_llm._torch.pyexecutor.py_executor.logger") as mock_logger:
+        _run_pad(stub)
+
+    assert mock_logger.warning.call_count == 0
+    # One routable request means the rank has real work; no dummy needed.
+    assert stub.add_dummy_calls == []
+
+
+def test_pad_still_warns_on_a_genuine_routable_surplus():
+    stub = _StubADPExecutor()
+    stub.active_requests = [
+        _make_adp_request(_STATE_GENERATION_IN_PROGRESS, request_id=1),
+        _make_adp_request(_STATE_GENERATION_IN_PROGRESS, request_id=2),
+    ]
+    stub.expected_num_active_requests = 1
+
+    with patch("tensorrt_llm._torch.pyexecutor.py_executor.logger") as mock_logger:
+        _run_pad(stub)
+
+    assert mock_logger.warning.call_count == 1
+    assert "exceeds expected_num_active_requests" in mock_logger.warning.call_args[0][0]
+
+
 def test_generic_disagg_adp_mixed_rank_states_stay_queueable():
     # The generic non-PP path must give both ranks a non-empty scheduled batch:
     # one rank schedules its real request, while the terminal-only rank
diff --git a/tests/unittest/_torch/executor/test_seq_slot_sizing.py b/tests/unittest/_torch/executor/test_seq_slot_sizing.py
index 7db2c6ed74ac..e24ab54963e4 100644
--- a/tests/unittest/_torch/executor/test_seq_slot_sizing.py
+++ b/tests/unittest/_torch/executor/test_seq_slot_sizing.py
@@ -1,57 +1,84 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
-"""Disaggregated attention-DP seq-slot sizing includes overlap headroom.
+"""Attention-DP seq-slot sizing includes overlap headroom.
 
 Under the overlap scheduler, requests finished in the previous iteration
 still hold their sequence slots when the next iteration's
-prepare_resources runs, while the V2 scheduler has already dropped them
-from its budget (no_schedule_after_state=GENERATION_TO_COMPLETE) and
-backfilled their seats. Transient slot demand is therefore
-2 * max_batch_size, regardless of whether speculative decoding is enabled.
-The headroom is selected from runtime topology rather than model architecture.
+prepare_resources runs, while the capacity scheduler has already dropped
+them from its budget (no_schedule_after_state=GENERATION_TO_COMPLETE) and
+backfilled their seats. Transient slot demand is therefore one extra
+micro-batch worth of slots, regardless of whether speculative decoding is
+enabled. The headroom is selected from runtime topology, not model
+architecture.
+
+Pipeline parallelism is out of scope: the pool is already sized by pp_size
+there, and the ADP router's retiring-request correction is not rank-consistent
+because only the last pipeline stage marks generation requests
+GENERATION_TO_COMPLETE.
 
 compute_max_num_sequences is the single sizing implementation used both
 for the executor's SeqSlotManager pool (create_py_executor_instance) and
-for the sampler state (create_torch_sampler_args).
+for the sampler state (create_torch_sampler_args); resolve_max_num_sequences
+is how a consumer obtains it without re-deriving it.
 """
 
+import inspect
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
 import pytest
 
 from tensorrt_llm._torch.pyexecutor._util import (
+    KvCacheCreator,
     compute_max_num_sequences,
     create_torch_sampler_args,
+    is_disagg_enabled,
+    resolve_max_num_sequences,
     should_enable_adp_dummy_fixes,
     should_enable_disagg_adp_overlap_headroom,
     should_enable_non_overlap_adp_forward_intent,
     should_enable_scheduler_aware_adp_dummy,
 )
-from tensorrt_llm.llmapi.llm_args import CacheTransceiverConfig
 from tensorrt_llm.mapping import Mapping
 
+_UCX = SimpleNamespace(backend="UCX")
+
+# (pp_size, enable_overlap_headroom, expected_factor)
+#
+# Disaggregation is absent from this table on purpose -- see
+# test_seat_pool_has_no_disagg_term.
 SIZING_CASES = [
-    # (pp_size, disable_overlap, enable_overlap_headroom, expected_factor)
-    (1, False, True, 2),
-    (1, False, False, 1),
-    (1, True, True, 1),
-    # Existing PP sizing is preserved regardless of the DSv4 opt-in.
-    (2, False, True, 2),
-    (4, False, True, 4),
-    (4, True, False, 4),
+    # No PP: the headroom buys one extra micro-batch worth of seats.
+    (1, False, 1),
+    (1, True, 2),
+    # PP sizes the pool by pipeline depth and ignores the headroom.
+    (4, False, 4),
+    (2, True, 2),
+    (4, True, 4),
 ]
 
 
 @pytest.mark.parametrize(
-    "enable_attention_dp,is_disagg,pp_size,disable_overlap,expected",
+    "enable_attention_dp,pp_size,cache_transceiver_config,disable_overlap,expected",
     [
-        (True, True, 1, False, True),
-        (False, True, 1, False, False),
-        (True, False, 1, False, False),
-        (True, True, 2, False, False),
-        (True, True, 1, True, False),
+        # Aggregated ADP with overlap on: nvbug 6627795 reproduced here, on a
+        # context-only run with no cache transceiver configured.
+        (True, 1, None, False, True),
+        (False, 1, None, False, False),
+        # Aggregated ADP with overlap off: teardown is in-line, no headroom.
+        (True, 1, None, True, False),
+        # Disagg needs the headroom even with overlap off: a request awaiting its
+        # KV transfer keeps its lease.
+        (True, 1, _UCX, True, True),
+        (True, 1, _UCX, False, True),
+        (False, 1, _UCX, False, False),
+        # Pipeline parallelism is out of scope in both directions.
+        (True, 2, None, False, False),
+        (True, 4, _UCX, False, False),
     ],
 )
 def test_disagg_adp_overlap_headroom_gate(
-    enable_attention_dp, is_disagg, pp_size, disable_overlap, expected
+    enable_attention_dp, pp_size, cache_transceiver_config, disable_overlap, expected
 ):
     mapping = Mapping(
         world_size=pp_size,
@@ -59,10 +86,11 @@ def test_disagg_adp_overlap_headroom_gate(
         pp_size=pp_size,
         enable_attention_dp=enable_attention_dp,
     )
-    cache_config = CacheTransceiverConfig(backend="NIXL") if is_disagg else None
 
     assert (
-        should_enable_disagg_adp_overlap_headroom(mapping, cache_config, disable_overlap)
+        should_enable_disagg_adp_overlap_headroom(
+            mapping, cache_transceiver_config, disable_overlap
+        )
         is expected
     )
 
@@ -101,11 +129,9 @@ def test_non_overlap_adp_forward_intent_scope(pp_size, disable_overlap, expected
     assert should_enable_non_overlap_adp_forward_intent(mapping, disable_overlap) is expected
 
 
-@pytest.mark.parametrize(
-    "pp_size,disable_overlap,enable_overlap_headroom,expected_factor", SIZING_CASES
-)
+@pytest.mark.parametrize("pp_size,enable_overlap_headroom,expected_factor", SIZING_CASES)
 def test_compute_max_num_sequences_scopes_overlap_headroom(
-    pp_size, disable_overlap, enable_overlap_headroom, expected_factor
+    pp_size, enable_overlap_headroom, expected_factor
 ):
     max_batch_size = 8
     mapping = Mapping(world_size=pp_size, tp_size=1, pp_size=pp_size)
@@ -113,22 +139,138 @@ def test_compute_max_num_sequences_scopes_overlap_headroom(
         compute_max_num_sequences(
             mapping,
             max_batch_size,
-            disable_overlap,
+            disable_overlap_scheduler=False,
             enable_overlap_headroom=enable_overlap_headroom,
         )
         == max_batch_size * expected_factor
     )
 
 
+def test_seat_pool_has_no_disagg_term():
+    """The disaggregation 2x reaches the seat pool only through the gate.
+
+    A request awaiting its KV transfer holds an *index* lease and no seat at all:
+    ``SeqSlotManager.prepare_resources`` skips ``DISAGG_GENERATION_INIT``
+    requests outright and only seats one once its transmission completes. So the
+    sizing function itself must not carry a disaggregation term -- the single
+    ``enable_overlap_headroom`` flag is the only way in.
+
+    Asserting on the signature rather than on a return value is deliberate: a
+    value test cannot distinguish "the parameter is gone" from "the parameter
+    defaults to False", and it is the parameter's *existence* that invites a
+    caller to propagate the factor.
+    """
+    assert "is_disagg" not in inspect.signature(compute_max_num_sequences).parameters
+    assert "is_disagg" not in inspect.signature(resolve_max_num_sequences).parameters
+
+
+@pytest.mark.parametrize(
+    "cache_transceiver_config,expected",
+    [
+        (None, False),
+        (SimpleNamespace(backend=None), False),
+        (SimpleNamespace(backend="UCX"), True),
+    ],
+)
+def test_is_disagg_enabled_is_the_single_definition(cache_transceiver_config, expected):
+    """One definition of "this is a disaggregated server".
+
+    The ``backend is not None`` test used to be inlined at each use site, which is
+    how a derived fact acquires copies that then disagree.
+    """
+    assert is_disagg_enabled(cache_transceiver_config) is expected
+
+
+@pytest.mark.parametrize(
+    "explicit,engine_seats,expected",
+    [
+        (24, 16, 24),  # an explicit value wins
+        (None, 16, 16),  # otherwise the engine's own pool
+        (None, None, 16),  # only then recompute, *with* the engine's gate
+    ],
+)
+def test_resolve_max_num_sequences_prefers_the_published_pool(explicit, engine_seats, expected):
+    """The fallback must never be able to undercut the pool it indexes.
+
+    The recomputing branch used to be the *first* branch and was called without
+    ``enable_overlap_headroom``, so a caller that omitted ``max_num_sequences``
+    silently sized the sampler and the executor's SeqSlotManager below the index
+    pool they share indices with. The third row is that branch, and it must still
+    land on the headroom value.
+    """
+    engine = SimpleNamespace(
+        max_num_seq_slots=engine_seats,
+        _enable_disagg_adp_overlap_headroom=True,
+    )
+    mapping = Mapping(world_size=1, tp_size=1, pp_size=1, enable_attention_dp=True)
+    llm_args = SimpleNamespace(disable_overlap_scheduler=False)
+
+    assert (
+        resolve_max_num_sequences(
+            engine,
+            mapping,
+            8,
+            llm_args,
+            max_num_sequences=explicit,
+        )
+        == expected
+    )
+
+
+def test_resolve_max_num_sequences_reads_llm_args_only_in_the_fallback():
+    """The two short-circuit branches must not touch ``llm_args`` at all.
+
+    Reading ``disable_overlap_scheduler`` at the *call site* made every caller
+    depend on a field only the third branch uses, which broke callers that hold a
+    lighter args object and pass ``max_num_sequences`` explicitly. An args object
+    that raises on attribute access is the only way to state that as a test:
+    asserting on the return value cannot distinguish "not used" from "used and
+    happened to agree".
+    """
+
+    class _Exploding:
+        def __getattr__(self, name):
+            raise AssertionError(f"llm_args.{name} read on a path that must not need it")
+
+    engine_with_pool = SimpleNamespace(max_num_seq_slots=16)
+    mapping = Mapping(world_size=1, tp_size=1, pp_size=1, enable_attention_dp=True)
+
+    # Branch 1: an explicit value wins, even with no pool published at all.
+    assert (
+        resolve_max_num_sequences(
+            SimpleNamespace(),
+            mapping,
+            8,
+            _Exploding(),
+            max_num_sequences=24,
+        )
+        == 24
+    )
+    # Branch 2: the engine's published pool.
+    assert resolve_max_num_sequences(engine_with_pool, mapping, 8, _Exploding()) == 16
+
+
+def test_sampler_args_require_the_resolved_pool():
+    """``max_num_sequences`` is required, and the raw material for re-deriving it
+    is gone from the signature.
+
+    ``create_torch_sampler_args`` used to default it by recomputing from
+    ``mapping``/``max_batch_size`` without the headroom gate, i.e. it could only
+    ever produce a number smaller than the slots the sampler indexes.
+    """
+    params = inspect.signature(create_torch_sampler_args).parameters
+
+    assert params["max_num_sequences"].default is inspect.Parameter.empty
+    assert "mapping" not in params
+    assert "max_batch_size" not in params
+
+
 @pytest.mark.parametrize("slot_factor", [1, 2])
 def test_sampler_uses_executor_slot_pool_capacity(slot_factor):
     max_batch_size = 8
-    mapping = Mapping(world_size=1, tp_size=1, pp_size=1)
     max_num_sequences = max_batch_size * slot_factor
     args = create_torch_sampler_args(
-        mapping,
         max_seq_len=1024,
-        max_batch_size=max_batch_size,
         speculative_config=None,
         max_beam_width=1,
         disable_overlap_scheduler=False,
@@ -137,3 +279,52 @@ def test_sampler_uses_executor_slot_pool_capacity(slot_factor):
         max_num_sequences=max_num_sequences,
     )
     assert args.max_num_sequences == max_num_sequences
+
+
+def _make_kv_cache_creator(disable_overlap_scheduler: bool) -> KvCacheCreator:
+    """Minimal creator whose only job is to reach _create_kv_cache_manager."""
+    c = object.__new__(KvCacheCreator)
+    c._mapping = Mapping(world_size=1, tp_size=1, pp_size=1)
+    c._kv_cache_config = Mock()
+    c._tokens_per_block = 32
+    c._max_seq_len = 1024
+    c._max_batch_size = 8
+    c._max_num_tokens = 8192
+    c._max_beam_width = 1
+    c._speculative_config = None
+    c._sparse_attention_config = None
+    c._kv_connector_manager = None
+    c._execution_stream = None
+    c._is_disagg = False
+    c._disable_overlap_scheduler = disable_overlap_scheduler
+    # Short-circuit the post-construction max_seq_len fixup.
+    c._skip_est = True
+    c._get_model_kv_cache_manager_cls = Mock(return_value=Mock())
+    c._should_create_separate_draft_kv_cache = Mock(return_value=False)
+    c._enable_kv_cache_stats = Mock(return_value=False)
+    return c
+
+
+@pytest.mark.parametrize("disable_overlap_scheduler", [False, True])
+def test_kv_cache_manager_receives_the_overlap_flag(disable_overlap_scheduler):
+    """The manager sizes its own index pool, but needs the overlap flag to do it.
+
+    The index pool must cover both the retiring cohort and its replacement when
+    attention DP runs with the overlap scheduler, otherwise ``_create_kv_cache``
+    silently defers admitted requests one at a time (nvbug 6627795). The flag is
+    passed rather than the seat-pool size so the manager keeps deriving its
+    capacity from ``max_batch_size * pp_size``, which is not comparable with a
+    seat pool that also carries the PP multiplier.
+    """
+    creator = _make_kv_cache_creator(disable_overlap_scheduler)
+    model_engine = SimpleNamespace(
+        model=SimpleNamespace(model_config=SimpleNamespace(is_generation=True)),
+    )
+
+    with patch(
+        "tensorrt_llm._torch.pyexecutor._util._create_kv_cache_manager",
+        return_value=None,
+    ) as create:
+        creator._create_kv_cache_manager(model_engine)
+
+    assert create.call_args.kwargs["disable_overlap_scheduler"] is disable_overlap_scheduler
diff --git a/tests/unittest/_torch/modeling/test_qwen4_exp_support.py b/tests/unittest/_torch/modeling/test_qwen4_exp_support.py
index 5da980b03c4c..cb98535d9942 100644
--- a/tests/unittest/_torch/modeling/test_qwen4_exp_support.py
+++ b/tests/unittest/_torch/modeling/test_qwen4_exp_support.py
@@ -1008,6 +1008,7 @@ def make_manager(
         hidden_size: int,
         max_num_requests: int,
         sa_manager: object = None,
+        num_seq_slots: Optional[int] = None,
     ) -> object:
         captured.update(
             config=config,
@@ -1015,6 +1016,7 @@ def make_manager(
             hidden_size=hidden_size,
             max_num_requests=max_num_requests,
             sa_manager=sa_manager,
+            num_seq_slots=num_seq_slots,
         )
         return captured
 
@@ -1044,6 +1046,9 @@ def make_manager(
     assert utils.get_spec_resource_manager(model_engine) is captured
     assert captured["hidden_size"] == 512
     assert captured["max_num_requests"] == 16
+    # This engine stub does not opt into the attention-DP overlap seq-slot
+    # headroom, so the manager must fall back to the max_num_requests sizing.
+    assert captured["num_seq_slots"] is None
 
 
 def test_logits_processor_borrows_target_mixer_but_mtp_head_owns_one() -> None:
diff --git a/tests/unittest/_torch/speculative/test_spec_slot_pool_sizing.py b/tests/unittest/_torch/speculative/test_spec_slot_pool_sizing.py
new file mode 100644
index 000000000000..86c18220b885
--- /dev/null
+++ b/tests/unittest/_torch/speculative/test_spec_slot_pool_sizing.py
@@ -0,0 +1,550 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Speculative-decoding state that is keyed by live-request identity must be
+sized by the sequence-slot pool, not by max_batch_size.
+
+Under the attention-DP overlap headroom the two differ by 2x
+(``compute_max_num_sequences``): a finished request holds its slot for one more
+iteration while its replacement is already admitted (nvbug-6627795). Two
+distinct families follow from that, and only the first needs the pool size:
+
+* keyed by ``py_seq_slot`` / a per-request ``SlotManager`` slot -- must span the
+  pool. ``SpecMetadata.num_seq_slots`` (draft_probs, full_draft_probs,
+  penalty_state), ``MTPHiddenStatesManager``'s hidden-state pools,
+  ``DynamicTreeSlotStorage``, ``Eagle3ResourceManager.slot_manager`` and the
+  ``SuffixAutomatonManager`` slot pool.
+* keyed by *batch position* -- ``max_num_requests`` is correct and deliberately
+  unchanged, because the micro-batch scheduler caps every forward at
+  max_batch_size (its ``no_schedule_after_state=GENERATION_TO_COMPLETE`` default
+  keeps the retiring requests out of the batch entirely). ``SpecTreeManager``'s
+  per-forward work buffers and ``batch_indices_cuda`` are in this family.
+"""
+
+import ast
+import inspect
+import textwrap
+import types
+
+import pytest
+import torch
+
+from tensorrt_llm._torch.pyexecutor.resource_manager import NoFreeSlotsError
+from tensorrt_llm._torch.speculative.eagle3 import (
+    Eagle3OneModelDynamicTreeResourceManager,
+    Eagle3ResourceManager,
+)
+from tensorrt_llm._torch.speculative.mtp import MTPHiddenStatesManager
+from tensorrt_llm._torch.speculative.mtp_dynamic_tree import MTPEagleDynamicTreeResourceManager
+from tensorrt_llm._torch.speculative.spec_tree_manager import SpecTreeManager
+from tensorrt_llm._torch.speculative.suffix_automaton import SAConfig, SuffixAutomatonManager
+from tensorrt_llm._torch.speculative.utils import (
+    _build_spec_metadata,
+    get_spec_drafter,
+    get_spec_metadata,
+    get_spec_resource_manager,
+)
+
+R, POOL = 8, 16  # max_batch_size, 2 * max_batch_size (overlap headroom)
+
+
+@pytest.mark.cpu_only
+def test_slot_pool_size_is_applied_centrally(monkeypatch):
+    """``get_spec_metadata`` stamps the pool size onto whatever mode was built.
+
+    This is the property that fixes the review finding: previously only the
+    MTP-eagle branch forwarded ``num_seq_slots``, so vanilla MTP, Eagle3
+    one-model, PARD, DFlash/DSpark and draft-target one-model all sized their
+    slot-indexed buffers at ``max_num_requests``. Applying it once at the single
+    exit point makes it impossible for a mode -- including a future one -- to be
+    missed, so the assertion deliberately does not name any mode.
+    """
+    built = types.SimpleNamespace()
+    monkeypatch.setattr(
+        "tensorrt_llm._torch.speculative.utils._build_spec_metadata", lambda *a, **k: built
+    )
+    spec_config = types.SimpleNamespace(enable_penalty=False)
+
+    out = get_spec_metadata(
+        spec_config,
+        model_config=object(),
+        max_num_requests=R,
+        max_num_tokens=128,
+        num_seq_slots=POOL,
+    )
+
+    assert out is built
+    assert out.num_seq_slots == POOL
+
+
+@pytest.mark.cpu_only
+def test_unknown_slot_pool_leaves_the_max_num_requests_fallback(monkeypatch):
+    """``num_seq_slots=None`` must not be written as a literal.
+
+    Both allocators resolve the pool as ``self.num_seq_slots or
+    self.max_num_requests``, so leaving the dataclass default (0) in place is how
+    a caller that does not know the pool size keeps the old sizing. Writing
+    ``None`` would raise in the ``+ 1`` scratch-row arithmetic instead.
+    """
+    built = types.SimpleNamespace()
+    monkeypatch.setattr(
+        "tensorrt_llm._torch.speculative.utils._build_spec_metadata", lambda *a, **k: built
+    )
+    spec_config = types.SimpleNamespace(enable_penalty=False)
+
+    get_spec_metadata(
+        spec_config,
+        model_config=object(),
+        max_num_requests=R,
+        max_num_tokens=128,
+        num_seq_slots=None,
+    )
+
+    assert not hasattr(built, "num_seq_slots")
+
+
+@pytest.mark.cpu_only
+def test_per_mode_builder_does_not_take_the_pool_size():
+    """Guard the central-application invariant structurally.
+
+    Re-plumbing ``num_seq_slots`` through the per-mode constructors is what let a
+    branch be forgotten in the first place; keep the builder free of it.
+    """
+    assert "num_seq_slots" not in inspect.signature(_build_spec_metadata).parameters
+
+
+def _mtp_config():
+    return types.SimpleNamespace(max_draft_len=2, use_relaxed_acceptance_for_thinking=True)
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="MTP hidden-state pools are CUDA tensors")
+@pytest.mark.parametrize(
+    "num_seq_slots,expected_pool",
+    [
+        (POOL, POOL + 1),
+        (None, R + 1),
+    ],
+)
+def test_mtp_hidden_states_pool_spans_the_slot_pool(num_seq_slots, expected_pool):
+    """The pool must cover every *resident* request, plus the CUDA-graph dummy.
+
+    ``add_slot`` runs on a request's first context chunk and the slot is only
+    returned by ``free_resources``, which the overlap scheduler defers -- so at
+    ``max_num_requests + 1`` the replacement request raises ``NoFreeSlotsError``.
+    ``None`` keeps the pre-existing sizing for callers that do not know the pool.
+    """
+    mgr = MTPHiddenStatesManager(
+        _mtp_config(), torch.float16, hidden_size=8, max_num_requests=R, num_seq_slots=num_seq_slots
+    )
+
+    assert mgr.slot_manager.max_num_requests == expected_pool
+    assert mgr.mtp_past_hidden_states_pool.shape[0] == expected_pool
+    assert mgr.mtp_past_tokens_pool.shape[0] == expected_pool
+    assert mgr.mtp_relaxed_delta_pool.shape[0] == expected_pool
+    # Batch-position state is unaffected: the forward batch is still capped at
+    # max_batch_size by the micro-batch scheduler.
+    assert mgr.get_max_resource_count() == R
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="MTP hidden-state pools are CUDA tensors")
+def test_mtp_slot_pool_survives_a_full_overlap_turnover():
+    """R retiring + R admitted must both hold slots at once.
+
+    This is the exact interleaving the overlap scheduler produces and the one
+    that used to exhaust the pool.
+    """
+    mgr = MTPHiddenStatesManager(
+        _mtp_config(), torch.float16, hidden_size=8, max_num_requests=R, num_seq_slots=POOL
+    )
+
+    retiring = [mgr.slot_manager.add_slot(rid) for rid in range(R)]
+    # Replacements are admitted before the deferred teardown frees the slots.
+    incoming = [mgr.slot_manager.add_slot(rid) for rid in range(R, 2 * R)]
+
+    assert len(set(retiring) | set(incoming)) == 2 * R
+    assert all(0 <= slot < POOL + 1 for slot in retiring + incoming)
+
+
+# ---------------------------------------------------------------------------
+# Resource managers. Unlike SpecMetadata there is no single exit point to stamp
+# the pool onto, so the plumbing is per-branch -- which is exactly how three
+# managers were missed in a row. The AST guard below makes forgetting a branch a
+# test failure instead of a runtime IndexError.
+# ---------------------------------------------------------------------------
+
+#: Managers that legitimately do not take a slot pool. Adding a name here must be
+#: a deliberate act with a reason, which is the point of the allow-list.
+_MANAGERS_WITHOUT_A_SLOT_POOL = {
+    # The n-gram pool is keyed by pattern, not by request identity, and NGRAM is
+    # absent from SpeculativeDecodingMode.support_overlap_scheduler(), so
+    # py_executor_creator forces the overlap scheduler off and the headroom can
+    # never apply.
+    "NGramPoolManager",
+    # Hidden-state export path; no per-request slot pool.
+    "SaveHiddenStatesResourceManager",
+}
+
+_MANAGERS_WITH_A_SLOT_POOL = (
+    MTPHiddenStatesManager,
+    MTPEagleDynamicTreeResourceManager,
+    Eagle3ResourceManager,
+    Eagle3OneModelDynamicTreeResourceManager,
+    SuffixAutomatonManager,
+    SpecTreeManager,
+)
+
+
+@pytest.mark.cpu_only
+def test_every_resource_manager_branch_forwards_the_slot_pool():
+    """Mechanical guard: no branch of ``get_spec_resource_manager`` may omit it.
+
+    ``num_seq_slots`` is computed once at the top of the function and then has to
+    reach every manager it builds. A new speculation mode -- or a new manager in
+    an existing mode's branch -- fails here rather than in production, where the
+    symptom is an out-of-range ``py_seq_slot`` write into a pool sized for
+    max_batch_size.
+    """
+    tree = ast.parse(textwrap.dedent(inspect.getsource(get_spec_resource_manager)))
+
+    missing = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
+        if name is None or not name.endswith("Manager") or name in _MANAGERS_WITHOUT_A_SLOT_POOL:
+            continue
+        if not any(kw.arg == "num_seq_slots" for kw in node.keywords):
+            missing.append(name)
+
+    assert not missing, (
+        f"get_spec_resource_manager builds {sorted(set(missing))} without forwarding "
+        "num_seq_slots; slot-keyed pools would be sized at max_batch_size. Either pass "
+        "it or justify the exemption in _MANAGERS_WITHOUT_A_SLOT_POOL."
+    )
+
+
+# ---------------------------------------------------------------------------
+# The drafter's own slot pool. Two-model speculation runs a *second*
+# SeqSlotManager (get_spec_drafter), and the guard above cannot see it: it walks
+# get_spec_resource_manager only, and matches names ending in "Manager" that take
+# a num_seq_slots *keyword*, while SeqSlotManager takes its size positionally.
+# That blind spot is why this pool stayed at max_batch_size while every pool
+# around it moved to the seat count.
+# ---------------------------------------------------------------------------
+
+
+def _drafter_engine(headroom: bool, seats: int = POOL):
+    """A stub target engine for get_spec_drafter's draft-target branch."""
+    spec_dec_mode = types.SimpleNamespace(
+        is_user_provided=lambda: False,
+        is_draft_target=lambda: True,
+        is_eagle3=lambda: False,
+        is_mtp_eagle=lambda: False,
+        is_ngram=lambda: False,
+    )
+    spec_config = types.SimpleNamespace(
+        spec_dec_mode=spec_dec_mode,
+        max_draft_len=2,
+        tokens_per_gen_step=3,
+        max_concurrency=None,
+        draft_len_schedule=None,
+    )
+    return types.SimpleNamespace(
+        spec_config=spec_config,
+        batch_size=R,
+        max_num_seq_slots=seats,
+        _enable_disagg_adp_overlap_headroom=headroom,
+    )
+
+
+def _drafter(headroom: bool, seats: int = POOL):
+    return get_spec_drafter(
+        _drafter_engine(headroom, seats),
+        draft_model_engine=object(),
+        sampler=object(),
+        spec_resource_manager=None,
+    )
+
+
+@pytest.mark.cpu_only
+@pytest.mark.parametrize("headroom,expected_pool", [(True, POOL), (False, R)])
+def test_draft_slot_pool_follows_the_target_seat_pool(headroom, expected_pool):
+    """The draft pool is sized by the *target* engine's seat count.
+
+    The indices it hands out address buffers sized by that seat count -- the
+    sampler shared with PyExecutor, the draft KV cache manager's IndexMapper, and
+    spec_resource_manager -- and ``_create_draft_request`` even carries the
+    target's ``py_seq_slot`` across as ``target_seq_slot``. Sizing this pool
+    independently is what makes the two disagree. Headroom off keeps
+    max_batch_size, so nothing changes for the other topologies.
+    """
+    assert _drafter(headroom).draft_seq_slot_manager.slot_manager.max_num_requests == expected_pool
+
+
+@pytest.mark.cpu_only
+def test_draft_slot_pool_survives_a_full_overlap_turnover():
+    """R retiring + R admitted, with the negative control that proves it bites.
+
+    ``cleanup_previous_draft_resources`` releases the previous draft batch's slots
+    a full iteration later (py_executor.py:5347), so with the headroom on both
+    cohorts hold draft slots at once. At max_batch_size the (R+1)-th lease raises
+    NoFreeSlotsError -- inside the draft loop, mid-iteration.
+    """
+    pool = _drafter(True).draft_seq_slot_manager.slot_manager
+    slots = [pool.add_slot(rid) for rid in range(2 * R)]
+    assert len(set(slots)) == 2 * R
+
+    starved = _drafter(False).draft_seq_slot_manager.slot_manager
+    for rid in range(R):
+        starved.add_slot(rid)
+    with pytest.raises(NoFreeSlotsError):
+        starved.add_slot(R)
+
+
+@pytest.mark.cpu_only
+def test_the_drafter_slot_pool_is_not_re_derived_from_max_batch_size():
+    """Structural guard, because the behavioural test above can be satisfied by
+    an accident of the stub: assert the source never passes a batch-size symbol
+    to SeqSlotManager. A new speculation mode that adds a second drafter branch
+    fails here rather than in a draft loop.
+    """
+    tree = ast.parse(textwrap.dedent(inspect.getsource(get_spec_drafter)))
+
+    offenders = []
+    seen = 0
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call) or getattr(node.func, "id", None) != "SeqSlotManager":
+            continue
+        seen += 1
+        for arg in list(node.args) + [kw.value for kw in node.keywords]:
+            if isinstance(arg, ast.Name) and arg.id in ("max_num_requests", "max_batch_size"):
+                offenders.append(arg.id)
+            elif isinstance(arg, ast.Attribute) and arg.attr in ("batch_size", "max_batch_size"):
+                offenders.append(arg.attr)
+
+    # Without this the guard passes vacuously the day the call is renamed away.
+    assert seen, "get_spec_drafter no longer builds a SeqSlotManager; retarget this guard"
+    assert not offenders, (
+        f"get_spec_drafter sizes its SeqSlotManager from {sorted(set(offenders))}; it must "
+        "follow the target engine's seat pool (seat_pool_or_none) or the draft lease for a "
+        "request seated above max_batch_size raises NoFreeSlotsError."
+    )
+
+
+@pytest.mark.cpu_only
+@pytest.mark.parametrize("manager", _MANAGERS_WITH_A_SLOT_POOL, ids=lambda m: m.__name__)
+def test_slot_pool_managers_accept_an_optional_pool_size(manager):
+    """The receiving end of the same contract, with ``None`` as the default.
+
+    ``None`` -- not ``max_num_requests`` -- has to be the default so that a caller
+    which does not know the pool (PP, no attention DP, overlap disabled) keeps the
+    established sizing without every call site having to restate it.
+    """
+    param = inspect.signature(manager.__init__).parameters.get("num_seq_slots")
+
+    assert param is not None, f"{manager.__name__} cannot be told its slot pool"
+    assert param.default is None, f"{manager.__name__} must default to None, got {param.default!r}"
+
+
+def _tree_manager(num_seq_slots):
+    return SpecTreeManager(
+        max_num_requests=R,
+        use_dynamic_tree=True,
+        max_total_draft_tokens=3,
+        max_draft_len=3,
+        eagle_choices=None,
+        dynamic_tree_max_topK=2,
+        num_seq_slots=num_seq_slots,
+    )
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="dynamic-tree slot storage is on CUDA")
+@pytest.mark.parametrize("num_seq_slots,expected_slots", [(POOL, POOL), (None, R)])
+def test_dynamic_tree_slot_storage_spans_the_slot_pool(num_seq_slots, expected_slots):
+    """``DynamicTreeSlotStorage`` is documented as indexed by ``py_seq_slot``.
+
+    It was nonetheless sized from ``num_trees`` (== max_batch_size), so the two
+    disagreed by 2x once the headroom was on. The dummy row sits one past the
+    pool, so every buffer is ``pool + 1`` deep.
+    """
+    storage = _tree_manager(num_seq_slots).slot_storage
+
+    assert storage.dummy_slot_id == expected_slots
+    for name in (
+        "has_tree",
+        "packed_mask",
+        "position_offsets",
+        "retrieve_index",
+        "retrieve_next_token",
+        "retrieve_next_sibling",
+    ):
+        assert getattr(storage, name).shape[0] == expected_slots + 1, name
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="dynamic-tree slot storage is on CUDA")
+def test_dynamic_tree_work_buffers_stay_at_max_batch_size():
+    """The other family must not be widened along with it.
+
+    ``num_trees`` indexes the build kernel's output by batch position, and the
+    micro-batch scheduler caps the forward at max_batch_size. Widening it would
+    waste memory quadratically in the tree dimensions for no benefit.
+    """
+    mgr = _tree_manager(POOL)
+
+    assert mgr.num_trees == R
+    assert mgr.retrieve_index.shape[0] == R
+    assert mgr.retrieve_next_token.shape[0] == R
+    assert mgr.retrieve_next_sibling.shape[0] == R
+    assert mgr.num_slots == POOL
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="dynamic-tree slot storage is on CUDA")
+def test_marking_a_high_slot_invalid_needs_the_pool():
+    """The concrete failure, plus a negative control that it was reachable.
+
+    ``Eagle3OneModelDynamicTreeResourceManager.free_resources`` calls
+    ``mark_invalid(request.py_seq_slot)``, and with the headroom on ``py_seq_slot``
+    ranges over the whole pool. Sized at max_batch_size the write is out of
+    range, so the second half of the assertion is what proves the first half is
+    not vacuous.
+    """
+    _tree_manager(POOL).slot_storage.mark_invalid(POOL - 1)
+
+    with pytest.raises(IndexError):
+        _tree_manager(None).slot_storage.mark_invalid(POOL - 1)
+
+
+def _eagle_config():
+    # Deliberately not an EagleDecodingConfig: that keeps max_total_draft_tokens
+    # on the max_draft_len branch and leaves spec_tree_manager unbuilt, so this
+    # exercises slot_manager sizing only.
+    return types.SimpleNamespace(
+        max_draft_len=2,
+        num_capture_layers=1,
+        use_relaxed_acceptance_for_thinking=True,
+    )
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="Eagle3 hidden states are CUDA tensors")
+@pytest.mark.parametrize("num_seq_slots,expected_pool", [(POOL, POOL + 1), (None, R + 1)])
+def test_eagle3_slot_manager_spans_the_slot_pool(num_seq_slots, expected_pool):
+    """``Eagle3ResourceManager`` sized its ``SlotManager`` from ``max_seq_len``.
+
+    That is a token count standing in for a slot count -- accidentally generous
+    for most configurations, but not for ``max_batch_size == max_seq_len``, where
+    the pool lands exactly one slot short of a full overlap turnover.
+    """
+    mgr = Eagle3ResourceManager(
+        _eagle_config(),
+        torch.float16,
+        hidden_size=8,
+        max_num_requests=R,
+        max_seq_len=4,
+        max_num_tokens=64,
+        num_seq_slots=num_seq_slots,
+    )
+
+    assert mgr.slot_manager.max_num_requests == expected_pool
+    assert mgr.relaxed_delta_pool.shape[0] == expected_pool
+    assert len(mgr.seq_lens) == expected_pool
+    assert len(mgr.start_indices) == expected_pool
+    # Batch-position state is untouched.
+    assert mgr.batch_indices_cuda.shape[0] == R
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="Eagle3 hidden states are CUDA tensors")
+def test_eagle3_keeps_the_max_seq_len_floor():
+    """Existing deployments must not shrink.
+
+    ``max_seq_len`` stays a floor so that every configuration where it already
+    exceeded the slot pool allocates exactly what it did before this change.
+    """
+    mgr = Eagle3ResourceManager(
+        _eagle_config(),
+        torch.float16,
+        hidden_size=8,
+        max_num_requests=R,
+        max_seq_len=1024,
+        max_num_tokens=64,
+        num_seq_slots=POOL,
+    )
+
+    assert mgr.slot_manager.max_num_requests == 1024 + 1
+
+
+def _sa_manager(num_seq_slots, **config_kwargs):
+    config = SAConfig(max_seq_len=1024, max_slots=R, **config_kwargs)
+    return SuffixAutomatonManager(config, R, 1024, num_seq_slots=num_seq_slots)
+
+
+@pytest.mark.cpu_only
+@pytest.mark.parametrize("num_seq_slots,expected_pool", [(POOL, POOL), (None, R)])
+def test_sa_pool_spans_the_slot_pool(num_seq_slots, expected_pool):
+    """SA slots are held for a request id's lifetime, so the pool follows it.
+
+    The dummy slot index is derived from ``pool_size``, so it moves with the pool
+    rather than colliding with a real slot.
+    """
+    mgr = _sa_manager(num_seq_slots)
+
+    assert mgr.pool_size == expected_pool
+    assert len(mgr._free_slots) == expected_pool
+    assert mgr._dummy_slot_index == expected_pool
+
+
+@pytest.mark.cpu_only
+def test_sa_pool_survives_a_full_overlap_turnover():
+    """2 * max_batch_size concurrent slots, with a negative control.
+
+    Without the pool the (max_batch_size + 1)-th allocation has nothing free and
+    nothing retained to evict, which is a hard ``RuntimeError`` mid-run.
+    """
+    mgr = _sa_manager(POOL)
+    slots = [mgr._allocate_slot() for _ in range(POOL)]
+    assert len(set(slots)) == POOL
+
+    starved = _sa_manager(None)
+    for _ in range(R):
+        starved._allocate_slot()
+    with pytest.raises(RuntimeError, match="No free or retained slots"):
+        starved._allocate_slot()
+
+
+@pytest.mark.cpu_only
+def test_an_explicit_sa_pool_is_a_floor_not_a_rejection():
+    """The seat count raises ``global_pool_size``; it never fails the run.
+
+    ``TorchLlmArgs.validate_speculative_config`` accepts any
+    ``global_pool_size >= max_batch_size``, so rejecting a value between
+    max_batch_size and the seat count would make merely enabling attention DP turn
+    an already-validated config into a startup error. The last assertion is the
+    negative control: without the headroom the configured value is used verbatim,
+    so this is a floor and not an unconditional bump.
+    """
+    assert _sa_manager(POOL, enable_global_pool=True, global_pool_size=64).pool_size == 64
+    assert _sa_manager(POOL, enable_global_pool=True, global_pool_size=R).pool_size == POOL
+    assert _sa_manager(None, enable_global_pool=True, global_pool_size=R).pool_size == R
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="dynamic-tree slot storage is on CUDA")
+def test_eagle3_one_model_dynamic_tree_forwards_the_slot_pool():
+    """End-to-end for the manager whose ``free_resources`` triggers the write."""
+    config = types.SimpleNamespace(
+        use_dynamic_tree=True,
+        max_draft_len=3,
+        tokens_per_gen_step=4,
+        eagle_choices=None,
+        dynamic_tree_max_topK=2,
+    )
+
+    mgr = Eagle3OneModelDynamicTreeResourceManager(config, R, num_seq_slots=POOL)
+
+    assert mgr.spec_tree_manager.slot_storage.dummy_slot_id == POOL
+    assert mgr.spec_tree_manager.num_trees == R
+    assert mgr.batch_indices_cuda.shape[0] == R
+    mgr.free_resources(types.SimpleNamespace(py_seq_slot=POOL - 1))
+
+
+if __name__ == "__main__":
+    import sys
+
+    sys.exit(pytest.main([__file__, "-v"]))