[BugFix][KVTransfer] Fix Mooncake split metadata group ids (vllm-project#11439)

## What this PR does / why we need it?

This PR fixes Mooncake KV split metadata generation for hybrid PD
disaggregation scenarios where Mooncake transfer groups do not have a
one-to-one mapping with KV cache manager groups.

PR vllm-project#10991 split Mooncake transfer groups by KV spec and preserved the
original KV cache manager group in `kv_cache_group_id`. PR vllm-project#10590 later
moved kernel block expansion into `_get_kv_split_metadata`, but that
path still indexed `local_block_ids` and `remote_block_ids` with the
transfer group id. For hybrid models such as Kimi, this can make a Mamba
or split attention transfer group index past the original KV cache group
block-id tuple.

The fix maps each transfer group back through `kv_cache_group_id` before
reading or writing split metadata block ids, covering both no-CP and
PCP/DCP paths.

## Does this PR introduce _any_ user-facing change?

No user-facing API change. It fixes internal Mooncake KV transfer
metadata generation for hybrid PD scenarios.

## How was this patch tested?
CI and Unit tests.
Unit tests:

```bash
uv run pytest -q tests/ut/kv_offload/test_mooncake_connector.py
```

Result: 93 passed, 22 subtests passed.


- vLLM version: v0.23.0
- vLLM main:
vllm-project/vllm@1f486d9

---------

Signed-off-by: zhuyixiang <zhuyixiang2014@163.com>

Author: Yuli-yx

tests/ut/kv_offload/test_mooncake_connector.py

@@ -2647,6 +2647,86 @@ def test_pd_disaggregated_hybrid_remote_pcp_splits_attention_and_final_mamba_sta
                     expected_finishes={0: 2, 1: 1},
                 )
 
+    def test_hybrid_no_cp_uses_kv_cache_group_ids_for_split_transfer_groups(self):
+        with patch.object(
+            self.vllm_config.kv_transfer_config,
+            "get_from_extra_config",
+            side_effect=lambda k, d=None: {
+                "prefill": {"tp_size": 4, "dp_size": 1, "pp_size": 1},
+                "decode": {"tp_size": 2, "dp_size": 1, "pp_size": 1},
+            }.get(k, d),
+        ):
+            self.vllm_config.scheduler_config.disable_hybrid_kv_cache_manager = False
+            self.vllm_config.model_config.is_deepseek_mla = False
+            self.vllm_config.model_config.hf_text_config.num_key_value_heads = 8
+            worker = MooncakeConnectorWorker(self.vllm_config, self.engine_id, MockKVCacheConfig())
+
+        worker._is_hma_required = True
+        worker.use_mla = False
+        worker.use_sparse = False
+        worker.num_key_value_heads = 8
+        worker.tp_size = 2
+        worker.tp_rank = 0
+        worker.pcp_size = 1
+        worker.dcp_size = 1
+        worker.pcp_rank = 0
+        worker.dcp_rank = 0
+        worker._decode_tp_size = 2
+        worker._prefill_tp_size = 4
+        worker._prefill_pp_size = 1
+        worker.side_channel_port = 5000
+        worker.handshake_port = worker.side_channel_port + worker.tp_rank
+        worker.local_remote_block_port_mapping = {}
+        worker.remote_port_send_num = {}
+        worker.block_size_scale = [[1], [1], [1]]
+        worker.kv_group2layeridx = {
+            0: (
+                {
+                    "kv_cache_spec_type": "FullAttentionSpec",
+                    "kv_cache_group_id": 0,
+                    "kv_cache_spec": {"num_kv_heads": 1},
+                },
+                [0],
+            ),
+            1: (
+                {
+                    "kv_cache_spec_type": "FullAttentionSpec",
+                    "kv_cache_group_id": 0,
+                    "kv_cache_spec": {"num_kv_heads": 8},
+                },
+                [1],
+            ),
+            2: (
+                {
+                    "kv_cache_spec_type": "MambaSpec",
+                    "kv_cache_group_id": 1,
+                },
+                [2],
+            ),
+        }
+
+        meta = types.SimpleNamespace(
+            remote_pcp_size=1,
+            remote_dcp_size=1,
+            remote_ptp_size=4,
+            remote_port=31000,
+            remote_block_ids=([50, 51, 52, 53], [60, 61, 62, 63]),
+            local_block_ids=([70, 71, 72, 73], [80, 81, 82, 83]),
+            num_external_tokens=4 * worker.block_size,
+            num_prompt_blocks=4,
+            num_computed_tokens=0,
+            remote_engine_id="remote_hybrid_split_transfer_groups",
+            remote_host="localhost",
+            remote_multi_nodes_meta_mapping={},
+            remote_block_size=16,
+        )
+
+        ports, local_ids, remote_ids = worker._get_kv_split_metadata("req_hybrid_split", cast(ReqMeta, meta))
+
+        self.assertEqual(len(ports), 1)
+        self.assertEqual(local_ids, [([70, 71, 72, 73], [80, 81, 82, 83])])
+        self.assertEqual(remote_ids, [([50, 51, 52, 53], [60, 61, 62, 63])])
+
     def test_get_tp_num_need_pulls(self):
         worker = MooncakeConnectorWorker(self.vllm_config, self.engine_id, MockKVCacheConfig())
         worker.num_key_value_heads = 8

vllm_ascend/distributed/kv_transfer/kv_p2p/mooncake_connector.py

@@ -2434,6 +2434,10 @@ def _group_compress_ratio(group_spec):
                     break
         return compress_ratio
 
+    @staticmethod
+    def _get_kv_cache_group_id(group_idx: int, group_spec: dict[str, Any]) -> int:
+        return group_spec.get("kv_cache_group_id", group_idx)
+
     def _get_kernel_block_ids(self, layer_indices, meta, group_idx, group_spec):
         """No-CP per-group block ids at kernel granularity: (local, remote).
 
@@ -2442,8 +2446,9 @@ def _get_kernel_block_ids(self, layer_indices, meta, group_idx, group_spec):
         kernels (already on D, located via num_computed_tokens), and trims both lists
         to the shorter one so remote/local stay aligned.
         """
+        kv_cache_group_id = self._get_kv_cache_group_id(group_idx, group_spec)
         if group_spec["kv_cache_spec_type"] == "MambaSpec":
-            return list(meta.local_block_ids[group_idx]), list(meta.remote_block_ids[group_idx])
+            return list(meta.local_block_ids[kv_cache_group_id]), list(meta.remote_block_ids[kv_cache_group_id])
 
         remote_block_size = meta.remote_block_size or self.block_size
 
@@ -2455,8 +2460,8 @@ def _get_kernel_block_ids(self, layer_indices, meta, group_idx, group_spec):
         )
 
         remote_scale = remote_block_size // kernel_size
-        kernel_local = self._expand_block_ids(list(meta.local_block_ids[group_idx]), local_scale)
-        kernel_remote = self._expand_block_ids(list(meta.remote_block_ids[group_idx]), remote_scale)
+        kernel_local = self._expand_block_ids(list(meta.local_block_ids[kv_cache_group_id]), local_scale)
+        kernel_remote = self._expand_block_ids(list(meta.remote_block_ids[kv_cache_group_id]), remote_scale)
         # Skip prefix-cached remote kernels (D-side already holds them). The token size of one
         # remote kernel is kernel_size * compress_ratio, so the number to skip is
         # num_computed_tokens // (kernel_size * compress_ratio).
@@ -2551,14 +2556,15 @@ def _get_kv_split_metadata(
             remote_handshake_port_list = [[x + meta.remote_port for x in chosen_rank_list]]
             # No CP: expand logical blocks into kernel blocks here so the transfer
             # stage consumes kernel-level ids directly (chunk_starts no longer needed).
-            local_block_ids: list = []
-            remote_block_ids: list = []
+            local_block_ids: list[list[int]] = [[] for _ in meta.local_block_ids]
+            remote_block_ids: list[list[int]] = [[] for _ in meta.remote_block_ids]
             for group_idx, (group_spec, layer_indices) in self.kv_group2layeridx.items():
                 local_kernel_block_ids, remote_kernel_block_ids = self._get_kernel_block_ids(
                     layer_indices, meta, group_idx, group_spec
                 )
-                local_block_ids.append(local_kernel_block_ids)
-                remote_block_ids.append(remote_kernel_block_ids)
+                kv_cache_group_id = self._get_kv_cache_group_id(group_idx, group_spec)
+                local_block_ids[kv_cache_group_id] = local_kernel_block_ids
+                remote_block_ids[kv_cache_group_id] = remote_kernel_block_ids
             local_block_ids_list = [tuple(local_block_ids) for _ in remote_handshake_port_list]
             remote_block_ids_list = [tuple(remote_block_ids) for _ in remote_handshake_port_list]
             return (