一次「缓存地址错配」的修复：
让 SFA indexer 在 DCP 下独立构建自己的元数据

    def build(
        self,
        common_prefix_len: int,
        common_attn_metadata: CommonAttentionMetadata,
        fast_build: bool = False,
        **kwargs,
    ) -> AscendSFAIndexerMetadata:
        # common_prefix_len / fast_build 保留仅为 API 兼容
        return self._build(
            common_attn_metadata,
            buffer_key=self._metadata_buffer_key(common_attn_metadata),
            use_cached_rope=True,
            # 投机解码会提前构建多份 step 元数据，每份都要独立 RoPE
            copy_rope=self.speculative_config is not None,
            **kwargs,
        )

    def _build_dcp_cache_metadata(
        self, common_attn_metadata, pcp_context, pcp_cache_group_idx, buffer_key,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        pcp_slot_mapping = None
        # PCP 开启 且 正在 prefill → PCP 顺序路径
        if self.use_pcp and pcp_context is not None and bool(pcp_context.global_batch.is_prefilling_np.any()):
            if pcp_cache_group_idx is None:
                raise RuntimeError("PCP+DCP indexer metadata requires the PCP cache-group index.")
            pcp_slot_mapping = self._build_pcp_ordered_slot_mapping(
                common_attn_metadata, pcp_context, pcp_cache_group_idx, buffer_key)
        # 常规路径：local 块表 → replicated 块表 → replicated slot
        num_reqs = common_attn_metadata.num_reqs
        dcp_block_table = get_sfa_dcp_local_block_table(
            common_attn_metadata.block_table_tensor, num_reqs, self.max_local_block_table_cols)
        block_table = self._build_block_table_replicated_view(
            dcp_block_table, common_attn_metadata.seq_lens, buffer_key)
        slot_mapping = self._build_slot_mapping_replicated_view(
            common_attn_metadata, block_table, buffer_key)
        if pcp_slot_mapping is not None:
            slot_mapping = pcp_slot_mapping
        return block_table, slot_mapping

def build_sfa_dcp_replicated_block_table(
    dcp_block_table, seq_lens, block_table_output,
    replicated_col_idx, dcp_size, blocks_per_physical_block,
):
    """Expand a local SFA block table into replicated-indexer addresses."""
    local_col_idx = (
        replicated_col_idx // (dcp_size * blocks_per_physical_block)
        * blocks_per_physical_block
        + replicated_col_idx % blocks_per_physical_block
    )
    rank_in_replicated_view = (replicated_col_idx // blocks_per_physical_block) % dcp_size

    local_logical_blocks = torch.index_select(dcp_block_table, 1, local_col_idx)
    if blocks_per_physical_block == 1:
        # [关键] 核心换算：本地物理块 → replicated 块号
        replicated_blocks = local_logical_blocks * dcp_size + rank_in_replicated_view
    else:
        local_sub_blocks = local_logical_blocks % blocks_per_physical_block
        local_physical_blocks = local_logical_blocks // blocks_per_physical_block
        replicated_blocks = (
            local_physical_blocks * dcp_size + rank_in_replicated_view
        ) * blocks_per_physical_block + local_sub_blocks

    valid_req_mask = ((seq_lens[: dcp_block_table.shape[0]].to(device=dcp_block_table.device) > 0)
                      .to(replicated_blocks.dtype).view(-1, 1))
    block_table_output.copy_(replicated_blocks * valid_req_mask)
    return block_table_output

def build_sfa_dcp_replicated_slot_mapping(
    common_attn_metadata, block_table_replicated_view,
    slot_mapping_output, replicated_view_block_size, device,
):
    """Build slots that address the physically replicated indexer cache."""
    num_reqs = common_attn_metadata.num_reqs
    num_actual_tokens = min(common_attn_metadata.num_actual_tokens, common_attn_metadata.num_input_tokens)
    num_query_tokens = int(common_attn_metadata.query_start_loc_cpu[num_reqs])
    slot_mapping_output.fill_(-1)                    # 默认占位：不可写
    if num_actual_tokens == 0:
        return slot_mapping_output
    query_lens = (common_attn_metadata.query_start_loc[1 : num_reqs + 1]
                  - common_attn_metadata.query_start_loc[:num_reqs])
    req_indices = torch.repeat_interleave(
        torch.arange(num_reqs, dtype=torch.int32, device=device),
        query_lens.to(device=device), output_size=num_query_tokens,
    )[:num_actual_tokens]
    ...
    logical_block_idx = positions // replicated_view_block_size
    block_offsets = positions % replicated_view_block_size
    block_table_indices = req_indices * block_table_replicated_view.shape[1] + logical_block_idx
    block_numbers = block_table_replicated_view.flatten()[block_table_indices]
    # [关键] slot = 块号 × 块大小 + 块内偏移
    slot_mapping_output[:num_actual_tokens] = block_numbers * replicated_view_block_size + block_offsets
    return slot_mapping_output

def build_pcp_ordered_slot_mapping(
    global_slot_mapping: torch.Tensor,
    pcp_context: Any,
    slot_mapping_buffer: torch.Tensor,
) -> torch.Tensor:
    """Write slots in PCP gather order into a caller-owned output buffer."""
    gather_idx = pcp_context.padded_gather_idx
    write_mask = pcp_context.gathered_kv_write_mask
    if gather_idx is None or write_mask is None:
        raise RuntimeError("PCP+DCP prefill requires the PCP gathered-token layout.")
    num_tokens = gather_idx.numel()
    if slot_mapping_buffer.shape[0] < num_tokens:
        raise RuntimeError(
            f"PCP+DCP indexer slot buffer is too small: "
            f"capacity={slot_mapping_buffer.shape[0]}, required={num_tokens}.")
    slot_mapping = slot_mapping_buffer[:num_tokens]
    # [关键] 按 gather 顺序 index_select，再掩掉不写的 token
    torch.index_select(global_slot_mapping, 0, gather_idx, out=slot_mapping)
    slot_mapping.masked_fill_(~write_mask, -1)
    return slot_mapping

    def _get_indexer_attn_metadata(self) -> Any | None:
        """只取 indexer 缓存层自己的元数据（由 indexer backend 构建）。"""
        if not self.runtime_has_indexer:
            return None
        own_prefix = self.indexer.k_cache.prefix
        prefixes = [own_prefix]
        kv_sharing_target = getattr(self, "kv_sharing_target_layer_name", None)
        if kv_sharing_target is not None:
            # 优先共享目标的 cache 视图，但保留草稿 indexer 自己的元数据
            target_base = kv_sharing_target.removesuffix(".attn")
            prefixes = [f"{target_base}.indexer.k_cache", own_prefix]
        forward_metadata = get_forward_context().attn_metadata
        indexer_metadata = None
        if isinstance(forward_metadata, dict):
            for prefix in prefixes:
                indexer_metadata = forward_metadata.get(prefix)
                if indexer_metadata is not None:
                    break
        if indexer_metadata is None:
            raise RuntimeError(
                f"No metadata was built for the indexer cache layer "
                f"prefixes={prefixes}. layer_name={self.layer_name}.")
        return indexer_metadata