Stabilize EAGLE draft cache hits under CP HiCache

The failing runs showed EAGLE accept length collapsing when draft cache-hit suffixes used the new partial-current splice path.  This keeps target partial-current reuse enabled, but returns EAGLE/NextN draft cache-hit suffixes to the previous full-materialize path with an explicit fallback warning until the draft splice path has value-level ETE proof.\n\nThe same change set also tightens the page-granular CP HiCache contract for scheduler-visible hits and makes the prefill-to-decode EAGLE handoff observable without cloning hot-path metadata.  Exact non-page CP hits are floored to a page boundary for new scheduling decisions, while internal unfinished-request refresh keeps its exact accounting.\n\nConstraint: CP shared KV and HiCache operate at page granularity; exposing token-precise CP tails to scheduler-visible cache hits can force non-page partial materialization.\nConstraint: EAGLE/NextN draft has only one executable layer, so draft prefetch and draft partial-current splice need a separate correctness contract from target layers.\nRejected: Keep draft partial-current splice enabled | remote logs correlate it with avg accept length around 0.068 and median 0.\nRejected: Clone decode metadata tensors on transfer | slot ownership until process_prebuilt consumes them avoids extra hot-path copies.\nConfidence: medium\nScope-risk: moderate\nDirective: Do not re-enable draft partial-current reuse without metadata/draft-KV value checks and ETE accept-length evidence.\nTested: g0034 container py_compile for touched modules.\nTested: g0034 container PYTHONPATH=python python -m pytest -q test/registered/unit/disaggregation/test_decode_queue_compaction.py test/registered/unit/mem_cache/test_cp_hicache_metadata.py test/registered/unit/mem_cache/test_cp_shared_kv_runtime.py -> 183 passed, 5 warnings, 2 subtests passed.\nNot-tested: Fresh ETE accept-length run after this exact commit; requires user-driven traffic restart.

python/sglang/srt/disaggregation/decode.py

@@ -42,6 +42,8 @@ from sglang.srt.disaggregation.utils import (
     ReqToMetadataIdxAllocator,
     TransferBackend,
     append_cp_draft_state_buffers,
+    eagle_accept_debug_should_log,
+    eagle_accept_debug_tensor_digest,
     get_kv_class,
     is_mla_backend,
     poll_and_all_reduce,
@@ -1172,6 +1174,25 @@ class DecodeTransferQueue:
             return True
 
         # Case 3: Success - commit the transfer
+        if (
+            not self.spec_algorithm.is_none()
+            and eagle_accept_debug_should_log("metadata_get")
+        ):
+            logger.warning(
+                "[EAGLE_ACCEPT_DEBUG][metadata_get] rid=%s room=%s idx=%s "
+                "output_id=%s cached_tokens=%s actual_room=%s topk_p=%s "
+                "topk_index=%s hidden=%s",
+                str(getattr(decode_req.req, "rid", ""))[:8],
+                decode_req.req.bootstrap_room,
+                idx,
+                int(output_id[0].item()),
+                int(cached_tokens[0].item()),
+                int(actual_room),
+                eagle_accept_debug_tensor_digest(output_topk_p[:1]),
+                eagle_accept_debug_tensor_digest(output_topk_index[:1]),
+                eagle_accept_debug_tensor_digest(output_hidden_states),
+            )
+
         decode_req.req.output_ids.append(output_id[0].item())
         decode_req.req.cached_tokens = cached_tokens[0].item()
         if not self.spec_algorithm.is_none():

python/sglang/srt/disaggregation/decode_schedule_batch_mixin.py

@@ -6,10 +6,16 @@ from typing import TYPE_CHECKING
 
 import torch
 
+from sglang.srt.disaggregation.utils import (
+    eagle_accept_debug_should_log,
+    eagle_accept_debug_tensor_digest,
+)
+from sglang.srt.environ import envs
 from sglang.srt.model_executor.forward_batch_info import CaptureHiddenMode, ForwardMode
 from sglang.srt.sampling.sampling_batch_info import SamplingBatchInfo
 
 logger = logging.getLogger(__name__)
+_EAGLE_ACCEPT_PREBUILT_DEBUG_COUNTER = 0
 
 if TYPE_CHECKING:
     from sglang.srt.managers.overlap_utils import FutureMap
@@ -42,6 +48,7 @@ class ScheduleBatchDisaggregationDecodeMixin:
         for i, req in enumerate(reqs):
             req_pool_indices.append(req.req_pool_idx)
 
+            pre_len = len(req.prefix_indices)
             chunk = self.req_to_token_pool.req_to_token[req.req_pool_idx][
                 : req.extend_input_len
             ]
@@ -51,7 +58,6 @@ class ScheduleBatchDisaggregationDecodeMixin:
             out_cache_loc[offset : offset + req.extend_input_len] = chunk
             offset += req.extend_input_len
 
-            pre_len = len(req.prefix_indices)
             seq_len = len(req.origin_input_ids) + max(0, len(req.output_ids) - 1)
             seq_lens.append(seq_len)
             if len(req.output_ids) == 0:
@@ -66,6 +72,35 @@ class ScheduleBatchDisaggregationDecodeMixin:
             pre_lens.append(pre_len)
             req.extend_logprob_start_len = 0
 
+            if envs.SGLANG_EAGLE_ACCEPT_DEBUG.get():
+                global _EAGLE_ACCEPT_PREBUILT_DEBUG_COUNTER
+                _EAGLE_ACCEPT_PREBUILT_DEBUG_COUNTER += 1
+                counter = _EAGLE_ACCEPT_PREBUILT_DEBUG_COUNTER
+                if (
+                    counter <= 16
+                    or counter % 256 == 0
+                    or (pre_len > 0 and (counter <= 128 or counter % 128 == 0))
+                ):
+                    logger.warning(
+                        "[EAGLE_ACCEPT_DEBUG][prebuilt_prepare] rid=%s "
+                        "pre_len=%s extend_input_len=%s fill_len=%s origin_len=%s "
+                        "output_len=%s seq_len=%s cached_tokens=%s "
+                        "req_pool_idx=%s out_chunk_start=%s expected_suffix_start=%s "
+                        "out_chunk_len=%s",
+                        str(getattr(req, "rid", ""))[:8],
+                        pre_len,
+                        req.extend_input_len,
+                        len(req.fill_ids),
+                        len(req.origin_input_ids),
+                        len(req.output_ids),
+                        seq_len,
+                        int(getattr(req, "cached_tokens", 0) or 0),
+                        req.req_pool_idx,
+                        0,
+                        pre_len,
+                        int(chunk.numel()),
+                    )
+
         extend_input_logprob_token_ids = None
 
         # Set fields
@@ -161,6 +196,29 @@ class ScheduleBatchDisaggregationDecodeMixin:
 
             hidden_states_list = [req.hidden_states_tensor for req in self.reqs]
             hidden_states = torch.stack(hidden_states_list, dim=0).to(self.device)
+            if eagle_accept_debug_should_log("prebuilt_state"):
+                req0 = self.reqs[0] if self.reqs else None
+                logger.warning(
+                    "[EAGLE_ACCEPT_DEBUG][prebuilt_state] rid=%s bs=%s "
+                    "num_states=%s output_ids=%s seq_lens=%s topk_p=%s "
+                    "topk_index=%s hidden=%s",
+                    str(getattr(req0, "rid", ""))[:8] if req0 is not None else None,
+                    len(self.reqs),
+                    num_states,
+                    self.output_ids[: min(4, self.output_ids.numel())].detach()
+                    .cpu()
+                    .tolist(),
+                    self.seq_lens[: min(4, self.seq_lens.numel())].detach()
+                    .cpu()
+                    .tolist(),
+                    eagle_accept_debug_tensor_digest(topk_p[: min(1, topk_p.shape[0])]),
+                    eagle_accept_debug_tensor_digest(
+                        topk_index[: min(1, topk_index.shape[0])]
+                    ),
+                    eagle_accept_debug_tensor_digest(
+                        hidden_states[: min(1, hidden_states.shape[0])]
+                    ),
+                )
 
             # local import to avoid circular import
             from sglang.srt.speculative.eagle_info import EagleDraftInput

python/sglang/srt/disaggregation/utils.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 import random
+import logging
 from collections import deque
 from contextlib import nullcontext
 from enum import Enum
@@ -14,6 +15,8 @@ import torch.distributed as dist
 from sglang.srt.environ import envs
 from sglang.srt.utils import is_npu
 
+logger = logging.getLogger(__name__)
+
 if TYPE_CHECKING:
     from sglang.srt.disaggregation.base.conn import KVArgs
     from sglang.srt.disaggregation.common.conn import (
@@ -84,6 +87,63 @@ def poll_and_all_reduce_attn_cp_tp_group(
 # Metadata Buffers
 #########################
 
+_EAGLE_ACCEPT_DEBUG_COUNTERS = {}
+
+
+def eagle_accept_debug_should_log(
+    key: str,
+    *,
+    first: int = 16,
+    every: int = 256,
+) -> bool:
+    if not envs.SGLANG_EAGLE_ACCEPT_DEBUG.get():
+        return False
+    count = _EAGLE_ACCEPT_DEBUG_COUNTERS.get(key, 0) + 1
+    _EAGLE_ACCEPT_DEBUG_COUNTERS[key] = count
+    return count <= first or (every > 0 and count % every == 0)
+
+
+def eagle_accept_debug_tensor_digest(tensor: Any, *, sample: int = 64) -> str:
+    """Small deterministic tensor summary for EAGLE handoff debugging.
+
+    This intentionally samples only a prefix.  It is enabled only under
+    SGLANG_EAGLE_ACCEPT_DEBUG and is for handoff equality checks, not for full
+    numerical validation.
+    """
+
+    if tensor is None:
+        return "None"
+    try:
+        t = torch.as_tensor(tensor)
+    except Exception as exc:  # pragma: no cover - defensive debug helper
+        return f"unavailable({type(exc).__name__})"
+    shape = tuple(t.shape)
+    dtype = str(t.dtype).replace("torch.", "")
+    numel = int(t.numel())
+    if numel == 0:
+        return f"shape={shape} dtype={dtype} numel=0"
+
+    flat = t.detach().reshape(-1)
+    sample_count = min(sample, numel)
+    try:
+        sample_cpu = flat[:sample_count].cpu()
+        head_cpu = sample_cpu[: min(8, sample_count)]
+        head = head_cpu.tolist()
+        if sample_cpu.is_floating_point():
+            checksum = float(sample_cpu.float().sum().item())
+            abs_checksum = float(sample_cpu.float().abs().sum().item())
+            return (
+                f"shape={shape} dtype={dtype} sample={sample_count} "
+                f"sum={checksum:.6g} abs={abs_checksum:.6g} head={head}"
+            )
+        checksum = int(sample_cpu.to(torch.int64).sum().item())
+        return (
+            f"shape={shape} dtype={dtype} sample={sample_count} "
+            f"sum={checksum} head={head}"
+        )
+    except Exception as exc:  # pragma: no cover - defensive debug helper
+        return f"shape={shape} dtype={dtype} numel={numel} digest_error={type(exc).__name__}"
+
 
 def append_cp_draft_state_buffers(
     kv_args: Any,
@@ -307,6 +367,27 @@ class MetadataBuffers:
             self.output_hidden_states[req.metadata_buffer_index].copy_(
                 req.hidden_states_tensor
             )
+            if eagle_accept_debug_should_log("metadata_set"):
+                logger.warning(
+                    "[EAGLE_ACCEPT_DEBUG][metadata_set] rid=%s room=%s idx=%s "
+                    "output_id=%s cached_tokens=%s topk=%s topk_p=%s "
+                    "topk_index=%s hidden=%s",
+                    str(getattr(req, "rid", ""))[:8],
+                    getattr(req, "bootstrap_room", None),
+                    req.metadata_buffer_index,
+                    req.output_ids[0] if req.output_ids else None,
+                    req.cached_tokens,
+                    topk,
+                    eagle_accept_debug_tensor_digest(
+                        self.output_topk_p[req.metadata_buffer_index, :topk]
+                    ),
+                    eagle_accept_debug_tensor_digest(
+                        self.output_topk_index[req.metadata_buffer_index, :topk]
+                    ),
+                    eagle_accept_debug_tensor_digest(
+                        self.output_hidden_states[req.metadata_buffer_index]
+                    ),
+                )
         # Store bootstrap_room for validation on decode side
         self.bootstrap_room[req.metadata_buffer_index, 0] = (
             req.bootstrap_room if req.bootstrap_room is not None else 0