Preserve CP shared-KV cache-hit diagnostics

Warm-cache GSM8K failures needed request-to-log correlation across scheduler, prefill transfer, and Mooncake CP filtering. The added diagnostics are gated by existing CP shared-KV debug envs and include bounded token signatures plus transfer page summaries so future debugging can identify whether a failed request hit L1, L2, or transfer truncation paths.\n\nThe findings document records the ruled-out hypotheses and the RAGGED current-row contract failure, preventing repeated log archaeology after context compaction.\n\nConstraint: Production hot paths must not emit these logs unless SGLANG_CP_SHARED_KV_BS_GT1_DEBUG or existing CP shared-KV debug gates are enabled.\nRejected: Add a new debug env | reuse of existing bs>1 debug gate avoids more runtime switches.\nRejected: Store full token ids in logs | bounded signatures are enough for correlation without huge logs.\nConfidence: medium\nScope-risk: narrow\nDirective: Keep these diagnostics gated and bounded; do not convert them to unconditional INFO logs.\nTested: Remote container py_compile as part of the RAGGED fix validation.\nNot-tested: Long-running production log-volume impact with debug enabled.

python/sglang/srt/disaggregation/mooncake/conn.py

@@ -64,6 +64,21 @@ def _cp_draft_shared_kv_debug(message: str, *args, limit: int = 64) -> None:
     logger.info("[CP_DRAFT_SHARED_KV] " + message, *args)
 
 
+def _cp_shared_kv_bs_gt1_mooncake_debug(
+    key: str, message: str, *args, limit: Optional[int] = None
+) -> None:
+    if not envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG.get():
+        return
+    if limit is None:
+        limit = int(envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG_LIMIT.get())
+    count_key = "bs_gt1:" + key
+    count = _CP_SHARED_DEBUG_COUNTS.get(count_key, 0)
+    if limit > 0 and count >= limit:
+        return
+    _CP_SHARED_DEBUG_COUNTS[count_key] = count + 1
+    logger.info("[CP_SHARED_KV_BS_GT1_DEBUG] event=%s " + message, key, *args)
+
+
 def _mooncake_transfer_stats_enabled() -> bool:
     return envs.SGLANG_DISAGGREGATION_TRANSFER_STATS.get()
 
@@ -1211,6 +1226,19 @@ class MooncakeKVManager(CommonKVManager):
                             _np_summary(chunked_dst_kv_indice),
                             kv_chunk.is_last_chunk,
                         )
+                        _cp_shared_kv_bs_gt1_mooncake_debug(
+                            "transfer_worker_kv",
+                            "cp_rank=%s room=%s prefill_pages=%s "
+                            "logical_positions=%s dst_pages=%s is_last=%s "
+                            "session=%s",
+                            self.attn_cp_rank,
+                            kv_chunk.room,
+                            _np_summary(kv_chunk.prefill_kv_indices),
+                            _np_summary(kv_chunk.logical_page_positions),
+                            _np_summary(chunked_dst_kv_indice),
+                            kv_chunk.is_last_chunk,
+                            req.mooncake_session_id,
+                        )
                         if envs.SGLANG_DEBUG_CP_SHARED_KV.get():
                             _cp_shared_debug_log(
                                 "transfer_worker_kv",
@@ -1326,6 +1354,18 @@ class MooncakeKVManager(CommonKVManager):
                                         ),
                                         _np_summary(dst_state_indices),
                                     )
+                                _cp_shared_kv_bs_gt1_mooncake_debug(
+                                    "transfer_worker_state",
+                                    "cp_rank=%s room=%s prefill_state_pages=%s "
+                                    "state_positions=%s dst_state_pages=%s "
+                                    "session=%s",
+                                    self.attn_cp_rank,
+                                    kv_chunk.room,
+                                    _np_summary(kv_chunk.state_indices),
+                                    _np_summary(kv_chunk.state_logical_page_positions),
+                                    _np_summary(dst_state_indices),
+                                    req.mooncake_session_id,
+                                )
                                 self.maybe_send_extra(
                                     req,
                                     kv_chunk.state_indices,
@@ -1706,6 +1746,22 @@ class MooncakeKVSender(CommonKVSender):
                 is_last_chunk,
                 getattr(self.kv_mgr.kv_args, "draft_kv_buffer_count", None),
             )
+            _cp_shared_kv_bs_gt1_mooncake_debug(
+                "sender_filter",
+                "cp_rank=%s room=%s page_start=%s orig_kv_pages=%s "
+                "filtered_kv_pages=%s kv_positions=%s orig_state_pages=%s "
+                "filtered_state_pages=%s state_positions=%s is_last=%s",
+                self.kv_mgr.attn_cp_rank,
+                self.bootstrap_room,
+                chunk_page_start,
+                _np_summary(orig_kv_indices),
+                _np_summary(kv_indices),
+                _np_summary(logical_page_positions),
+                _np_summary(orig_state_indices),
+                _np_summary(state_indices),
+                _np_summary(state_logical_page_positions),
+                is_last_chunk,
+            )
         # Special handling for cp
         elif self.kv_mgr.enable_all_cp_ranks_for_transfer:
             kv_indices, index_slice = filter_kv_indices_for_cp_rank(

python/sglang/srt/disaggregation/prefill.py

@@ -43,6 +43,7 @@ from sglang.srt.disaggregation.utils import (
     kv_to_page_num,
     poll_and_all_reduce_attn_cp_tp_group,
     prepare_abort,
+    token_ids_debug_signature,
 )
 from sglang.srt.managers.schedule_batch import (
     FINISH_ABORT,
@@ -1304,7 +1305,7 @@ class SchedulerDisaggregationPrefillMixin:
                 "rid=%s room=%s start_idx=%s end_idx=%s last_chunk=%s "
                 "page_size=%s pages=%s state_pages=%s prefix_len=%s "
                 "host_hit_length=%s extend_input_len=%s fill_len=%s "
-                "origin_input_len=%s has_draft_pool=%s",
+                "origin_input_len=%s has_draft_pool=%s origin_sig=%s fill_sig=%s",
                 req.rid,
                 req.bootstrap_room,
                 start_idx,
@@ -1319,6 +1320,8 @@ class SchedulerDisaggregationPrefillMixin:
                 len(req.fill_ids),
                 len(req.origin_input_ids),
                 has_draft_pool,
+                token_ids_debug_signature(getattr(req, "origin_input_ids", None)),
+                token_ids_debug_signature(getattr(req, "fill_ids", None)),
             )
         if has_draft_pool and draft_prefix_overlap > 0:
             _cp_draft_shared_kv_debug(

python/sglang/srt/disaggregation/utils.py

@@ -67,6 +67,31 @@ def _cp_shared_kv_poll_debug_enabled() -> bool:
     return envs.SGLANG_CP_SHARED_KV_BS_GT1_DEBUG.get()
 
 
+def token_ids_debug_signature(token_ids: Any, *, sample: int = 6) -> str:
+    """Return a stable, bounded signature for request-token correlation logs."""
+
+    if token_ids is None:
+        return "None"
+    if isinstance(token_ids, torch.Tensor):
+        values = token_ids.detach().reshape(-1).cpu().tolist()
+    elif isinstance(token_ids, np.ndarray):
+        values = token_ids.reshape(-1).tolist()
+    else:
+        values = list(token_ids)
+
+    mask = (1 << 63) - 1
+    digest = 1469598103934665603
+    for value in values:
+        item = int(value) & 0xFFFFFFFFFFFFFFFF
+        for shift in (0, 8, 16, 24, 32, 40, 48, 56):
+            digest ^= (item >> shift) & 0xFF
+            digest = (digest * 1099511628211) & mask
+
+    head = values[:sample]
+    tail = values[-sample:] if len(values) > sample else values[:]
+    return f"len={len(values)} hash={digest} head={head} tail={tail}"
+
+
 def _poll_queue_debug_hash(debug_ids: Optional[list[str]]) -> int:
     if not debug_ids:
         return 0

python/sglang/srt/managers/scheduler.py

@@ -60,6 +60,7 @@ from sglang.srt.disaggregation.utils import (
     ReqToMetadataIdxAllocator,
     TransferBackend,
     prepare_abort,
+    token_ids_debug_signature,
 )
 from sglang.srt.distributed import get_pp_group, get_world_group
 from sglang.srt.distributed.parallel_state import get_tp_group
@@ -2584,7 +2585,7 @@ class Scheduler(
                 "scheduler_prefill_batch",
                 "bs=%s extend_lens=%s prefix_lens=%s seq_lens=%s "
                 "out_cache_tokens=%s chunked_req=%s enable_bs_gt1=%s "
-                "max_batch_reqs=%s max_total_extend=%s",
+                "max_batch_reqs=%s max_total_extend=%s rids=%s origin_sigs=%s",
                 len(can_run_list),
                 list(getattr(new_batch, "extend_lens", []) or []),
                 list(getattr(new_batch, "prefix_lens", []) or []),
@@ -2603,6 +2604,11 @@ class Scheduler(
                     "cp_shared_kv_prefill_max_total_extend_tokens",
                     None,
                 ),
+                [getattr(req, "rid", None) for req in can_run_list],
+                [
+                    token_ids_debug_signature(getattr(req, "origin_input_ids", None))
+                    for req in can_run_list
+                ],
             )
 
         # Record prefill stats for logging after forward