diff --git a/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_implementation_plan_zh.md b/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_implementation_plan_zh.md
index 39017794d..a3012e440 100644
--- a/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_implementation_plan_zh.md
+++ b/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_implementation_plan_zh.md
@@ -9,6 +9,7 @@
 **当前约束：** 先保证 target model 同步路径正确，再恢复 current/partial-current reuse，再处理 EAGLE/draft，最后打开 MLA/index L1 prefetch。不能通过删除 `batch_size != 1` guard 来“强行开启”。
 
 - 并行派工版计划：`docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_parallel_workstreams_zh.md`。
+- W3/W4 target sync 细化文档：`docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_w3_w4_plan_zh.md`。
 
 ---
 
diff --git a/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_w3_w4_plan_zh.md b/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_w3_w4_plan_zh.md
new file mode 100644
index 000000000..990a76697
--- /dev/null
+++ b/docs/advanced_features/nsa_prefill_cp_shared_kv_bs_gt1_w3_w4_plan_zh.md
@@ -0,0 +1,637 @@
+# NSA Prefill CP Shared-KV bs>1 W3/W4 实现文档
+
+> 日期：2026-06-03
+> 分支：`cjy-cp-refactor`
+> 当前基线：`e4cf8d18b`
+> 范围：W3 `local out_cache_loc + direct write`，W4 `target index/top-k sync correctness`。
+
+> **命名说明：** 本文的 W4 指顺序实现计划中的 **Phase 4 target index/top-k sync correctness**。并行派工文档里的 W4 是 current/partial-current reuse；那部分在本文中仍视为后续阶段，不在本轮实现范围内。
+
+## 0. 目标和非目标
+
+目标是在 **不等待 W2 allocator 最终完成** 的前提下，先把 target model 的 W3/W4 runtime consumer 做成 batch-aware：
+
+1. bs>1 的 CP shared-KV direct write 使用每个 request 独立的 page-aligned split，再按 request order 拼接本 rank local rows。
+2. MLA KV direct write 和 index KV direct write 共享同一个 ForwardBatch 级 page/local-loc plan。
+3. target index/top-k sync path 使用 per-request metadata，不能把 batch flatten 成一条长序列。
+4. W4 第一版只做 sync correctness，不启用 bs>1 current reuse、partial-current reuse、L1 prefetch、draft/EAGLE。
+
+非目标：
+
+- 不实现 W2 owner-lane allocator；W3 单测使用 synthetic owner-valid `out_cache_loc`。
+- 不实现 bs>1 current/partial-current reuse。
+- 不实现 batched MLA/index prefetch。
+- 不实现 draft/EAGLE bs>1。
+- 不新增 collective 来同步 batch plan。
+
+## 1. 必须保持的核心语义
+
+CP shared-KV 下 page 相关规划是 **request/batch 级别**，不是 layer 级别：
+
+```text
+同一个 ForwardBatch 内：
+  logical page id / owner / page table / local loc / physical remap 不随 layer 变化
+
+每个 layer 内：
+  只有 KV bytes、index K/scale、materialized dense buffer 内容、topk/logits 变化
+```
+
+因此 W3/W4 的实现不能每层重新推导 page plan。应该在 `ForwardBatch` 上缓存：
+
+- local logical `out_cache_loc`；
+- local physical `out_cache_loc`；
+- paged slot remap；
+- request-local prev/next segment offsets；
+- request-local page/table descriptor。
+
+当前已有缓存字段：
+
+- `ForwardBatch.cp_local_out_cache_loc` 和 `cp_local_physical_out_cache_loc` 已在 `forward_batch_info.py:427-428`。
+- `ForwardBatch.cp_shared_kv_paged_slot_remap_key/remap` 已在 `forward_batch_info.py:432-433`。
+- `get_or_build_shared_paged_buffer_slot_remap()` 已在 `cp_shared_kv_runtime.py:2486-2520` 按 key 缓存 paged remap。
+
+W3/W4 的实现原则：
+
+```text
+prepare/batch plan 阶段确定 page 和 segment。
+direct write/top-k/materialize 只消费 plan。
+layer_id 只选择 layer buffer，不参与 page planning。
+```
+
+bs>1 的 padding 合同：
+
+```text
+不把短 request pad 到 batch 内最长 request 的长度。
+
+每个 request 只独立 pad 到 page_size 边界：
+  valid_tokens=100, page_size=64 -> padded_tokens=128
+
+ForwardBatch 内计算输入仍是 sum(request_extend_lens) 的 packed/ragged tokens。
+request 边界由 request_extend_lens / request_split_lists / cu_seqlens 保留。
+如果某条 fast path 需要等长，第一版应改为 segmented/per-request descriptor，
+不能通过 max-length padding 掩盖接口缺陷。
+```
+
+## 2. 已确认的代码事实
+
+### C1. W1 batch metadata 已经存在
+
+`CPSharedKVBatchPlan` 在 `nsa/utils.py:280-312`，包含：
+
+- `request_extend_lens`
+- `request_prefix_lens`
+- `request_padded_pages/tokens`
+- `request_split_lists`
+- `request_zigzag_indices`
+- `request_kv_len_prev/next`
+- `request_actual_seq_q_prev/next`
+- `request_rank_local_offsets`
+- `flat_*` views
+
+`build_batch_page_aligned_in_seq_split_plan()` 在 `nsa/utils.py:335-490`，每个 request 独立 page-align split，再 flatten。
+
+`prepare_input_dp_with_cp_dsa()` 在 `nsa/utils.py:1579-1646`，当 CP shared-KV 且 `len(extend_seq_lens_cpu) > 1` 时会构造 batch plan，并通过 `_build_batch_metadata_from_plan()` 写入 `NSAContextParallelMetadata`。
+
+### C2. batch split helper 已经存在，但 W3 local loc 尚未完全使用
+
+`split_tensor_by_cp_batch_plan()` 在 `nsa/utils.py:507-560`，会按 `request_extend_lens` 切 request，再按每个 request 的 `request_split_lists/request_zigzag_indices` 拼本 rank local rows。
+
+`cp_split_and_rebuild_data()` / `cp_split_and_rebuild_1d()` 已在 `nsa/utils.py:951` 附近 dispatch 到 `_cp_split_and_rebuild_batch_in_seq()`。
+
+### C3. 当前 W3 最大疏漏：`get_cp_shared_kv_local_out_cache_loc()` 仍用 scalar split 做长度检查
+
+`get_cp_shared_kv_local_out_cache_loc()` 在 `nsa/utils.py:1039-1124`。
+
+当前代码在 `nsa/utils.py:1093-1102`：
+
+```python
+split_tokens = sum(int(x) for x in metadata.split_list)
+out_cache_tokens = int(out_cache_loc.numel())
+if split_tokens != out_cache_tokens:
+    ...
+    return None
+```
+
+bs>1 下 `metadata.split_list` 是 `_build_batch_metadata_from_plan()` 写入的 **first request scalar compatibility field**，见 `nsa/utils.py:728-741`。因此这个检查会错误地把 bs>1 batch 判定为 mismatch，导致 direct write fallback。
+
+修复原则：
+
+```text
+if batch_plan.batch_size > 1:
+    expected = sum(batch_plan.request_extend_lens)
+else:
+    expected = sum(metadata.split_list)
+```
+
+然后仍然使用 `cp_split_and_rebuild_1d()` / `split_tensor_by_cp_batch_plan()` 得到 local logical loc。
+
+### C4. 当前 physical remap 已经是 ForwardBatch 级缓存
+
+`get_cp_shared_kv_local_physical_out_cache_loc()` 在 `nsa/utils.py:1127-1173`，注释已经写明 logical loc 是 batch-scoped, not layer-scoped，并缓存到 `forward_batch.cp_local_physical_out_cache_loc`。
+
+W3 需要保留这个设计。不要把 physical remap 做成 per-layer。
+
+### C5. MLA direct write 主要依赖 local loc helper
+
+`_maybe_write_cp_shared_local_mla_kv()` 在 `forward_mla.py:551-623`。
+
+流程：
+
+1. 通过 `get_cp_shared_kv_local_out_cache_loc()` 取得 local logical loc。
+2. 校验 `k_nope/k_pe.shape[0] == local_out_cache_loc.numel()`。
+3. 优先 `try_tai_fused_mla_store(...)`。
+4. 否则用 `get_cp_shared_kv_local_physical_out_cache_loc()` 后 `set_mla_kv_buffer()`。
+
+因此 W3 对 MLA 的主要实现面是 local loc helper 和测试。
+
+### C6. index direct write 失败后目前会 silent 落回 legacy store
+
+`_store_cp_shared_local_index_k_cache()` 在 `nsa_indexer.py:1606-1652`。
+
+如果 local loc 不可用或 shape mismatch，它返回 `False`。调用点在 `nsa_indexer.py:1720-1731`、`1745-1756`、`1760-1771` 会继续调用 `_store_index_k_cache()`。
+
+`_store_index_k_cache()` 如果未传 `out_loc_override`，会走 `_filter_shared_index_write()`，后者在 `nsa_indexer.py:535-579` 直接基于完整 `forward_batch.out_cache_loc` filter owned logical loc。
+
+这对历史 bs=1 是兼容 fallback，但对当前 CP shared-KV/page-aligned/direct-write 合同不合适：W3 支持后，bs=1/bs>1 都不应 silent fallback 到 legacy filter path。否则即使 allocator/page plan 有问题，也可能被 fallback 掩盖。
+
+W3 要求：
+
+```text
+CP shared-KV 合同内 + local loc unavailable / owner mismatch / shape mismatch:
+  index/MLA direct write 必须 fail-fast
+
+非 CP shared-KV 路由:
+  可以返回 False 走原路径；这不是 CP shared-KV fallback
+```
+
+### C7. 当前 W4 `_get_topk_in_seq_cp_pair()` 是单请求
+
+`_get_topk_in_seq_cp_pair()` 在 `nsa_indexer.py:1371-1455`。
+
+它读取 scalar：
+
+- `metadata.kv_len_prev`
+- `metadata.kv_len_next`
+- `metadata.actual_seq_q_prev`
+- `metadata.actual_seq_q_next`
+
+并用 `split_in_seq_cp_local_pair()` 把整个 local tensor 切成 prev/next 两段。
+
+bs>1 下 local tensor 实际应是：
+
+```text
+req0_prev, req0_next, req1_prev, req1_next, ...
+```
+
+不能用 first request 的 scalar lengths 切整批。
+
+### C8. `_get_topk_ragged_with_cp()` 的非 cp_index 分支硬编码 batch 0
+
+`_get_topk_ragged_with_cp()` 在 `nsa_indexer.py:1087-1369`。
+
+非 `cp_index` 分支在 `nsa_indexer.py:1224-1232` 使用：
+
+```python
+forward_batch.seq_lens_cpu[0]
+forward_batch.extend_seq_lens_cpu[0]
+```
+
+并在 `nsa_indexer.py:1254-1287` 使用：
+
+```python
+block_tables[0]
+```
+
+W4 如果想复用这个函数处理 bs>1 的每个 request，就必须新增 `batch_idx: int = 0` 参数，并用 `block_tables[batch_idx]`、`seq_lens_cpu[batch_idx]`、`extend_seq_lens_cpu[batch_idx]`。
+
+### C9. 不应使用现有 `cp_index` 分支作为 W4 第一版
+
+`_get_topk_ragged_with_cp()` 的 `cp_index` 分支在 `nsa_indexer.py:1155-1222`，源码注释明确写着 `TODO Multi-batch support has accuracy issues`。
+
+W4 第一版为了 correctness 应该避开该分支，采用 per-request/per-segment 同步调用：
+
+```text
+for req_id:
+  call prev segment
+  call next segment
+```
+
+后续性能优化可以再把多个 segment 合并成 batched top-k descriptor。
+
+### C10. W4 current/partial-current reuse 仍是单请求合同
+
+`_maybe_materialize_shared_index_buffer()` 在 `nsa_indexer.py:309-517`。
+
+当 `current_index_kv is not None` 时，它要求：
+
+```python
+len(prefix_lens_cpu) == 1
+positive page-aligned prefix
+```
+
+见 `nsa_indexer.py:341-355`。
+
+因此 W4 第一版遇到 bs>1 + `current_index_kv is not None` 必须 fail-fast：
+
+```text
+[CP_SHARED_KV_FAIL_FAST][batch_gt1_index_current_reuse_unsupported]
+```
+
+不要在 W4 里临时拼 current reuse，否则会和 W5 的 partial/current reuse 工作混在一起。
+
+### C11. `_build_batch_metadata_from_plan()` 的 batch cu tensor 不能直接传给单 segment top-k
+
+`_build_batch_metadata_from_plan()` 在 `nsa/utils.py:832-840` 构造：
+
+```python
+request_actual_seq_q_prev_cu_tensor = [0] + cumsum(request_actual_seq_q_prev)
+request_actual_seq_q_next_cu_tensor = [0] + cumsum(request_actual_seq_q_next)
+```
+
+这是 batch-level cumulative tensor，不是单个 request segment 的 `[0, segment_len]`。
+
+W4 的 per-request/per-segment top-k 调用应该传：
+
+```text
+actual_seq_q_tensor = tensor([segment_len])
+actual_seq_q_cu_tensor = tensor([0, segment_len])
+```
+
+第一版可以每次构造小 tensor；后续再优化为 metadata 预构造或 TAI descriptor。
+
+## 3. W3 设计：batch-aware local out_cache_loc + direct write
+
+### 3.1 输入合同
+
+W3 依赖以下输入：
+
+- `forward_batch.uses_cp_shared_kv == True`
+- `forward_batch.cp_shared_kv_layout is not None`
+- `forward_batch.nsa_cp_metadata` 包含 batch plan 或 batch fields
+- `forward_batch.out_cache_loc` 是按 request order flatten 的 **valid-token loc view**
+
+关于 W2 allocator 的交接：
+
+```text
+W3 第一版假设 out_cache_loc.numel() == sum(request_extend_lens)
+```
+
+如果 W2 allocator 以后返回 padded-token loc，则 W2 必须同时提供 valid view，或者 W3 增加 `request_valid_out_cache_loc` view；不能让 direct write 写 padded tail rows。
+
+### 3.2 输出合同
+
+`get_cp_shared_kv_local_out_cache_loc(forward_batch)` 返回：
+
+```text
+local logical loc = concat(local(req0), local(req1), ...)
+```
+
+其中每个 `local(req)` 按该 request 的 `request_zigzag_indices[req]` 取两个 in-seq local segment。
+
+`get_cp_shared_kv_local_physical_out_cache_loc(forward_batch)` 返回：
+
+```text
+layout.logical_locs_to_physical(local logical loc)
+```
+
+并在 ForwardBatch 上缓存。
+
+### 3.3 实现步骤
+
+#### W3-S1：新增 batch-local-loc helper
+
+建议在 `nsa/utils.py` 增加轻量 helper：
+
+```python
+def _get_cp_shared_kv_expected_out_cache_tokens(metadata) -> int:
+    plan = get_cp_shared_kv_batch_plan(...)
+    if plan is not None and plan.batch_size > 1:
+        return sum(plan.request_extend_lens)
+    return sum(metadata.split_list)
+```
+
+也可以直接写在 `get_cp_shared_kv_local_out_cache_loc()` 里，但 helper 更容易测。
+
+#### W3-S2：修改 length check
+
+将 scalar-only 检查替换成：
+
+```text
+batch plan exists:
+  expected_tokens = sum(request_extend_lens)
+  log reason = batch_out_cache_len_mismatch
+else:
+  expected_tokens = sum(split_list)
+```
+
+#### W3-S3：使用 batch split helper 生成 local logical loc
+
+继续调用：
+
+```python
+local_out_cache_loc = cp_split_and_rebuild_1d(forward_batch, out_cache_loc.contiguous())
+```
+
+因为 `cp_split_and_rebuild_1d()` 已经会在 `metadata.batch_size > 1` 时进入 `split_tensor_by_cp_batch_plan()`。
+
+#### W3-S4：保留 owner validation
+
+保留：
+
+```python
+valid_locs = local_out_cache_loc[local_out_cache_loc > 0]
+layout.owned_by_this_rank(valid_locs)
+```
+
+这个校验是 W2 allocator 和 W3 consumer 的关键集成检查。
+
+#### W3-S5：index direct-write 不允许 silent fallback
+
+在 `_store_cp_shared_local_index_k_cache()` 中，当：
+
+```text
+forward_batch.uses_cp_shared_kv
+local_out_loc is None or shape mismatch
+```
+
+必须 fail-fast，而不是返回 `False` 进入 `_store_index_k_cache()` legacy path。bs=1 和 bs>1 一致；非 CP shared-KV 可以继续返回 `False` 走原路径。
+
+#### W3-S6：MLA direct-write 不允许 silent fallback
+
+`_maybe_write_cp_shared_local_mla_kv()` 当前返回 `False` 后，前向会触发 legacy rebuild/materialize。CP shared-KV 合同内如果 local loc 缺失或 shape mismatch，应按同样规则 fail-fast。
+
+建议第一版：
+
+```text
+bs>1 + local loc unavailable -> RuntimeError [CP_SHARED_KV_FAIL_FAST][batch_gt1_mla_direct_write_unavailable]
+bs>1 + shape mismatch -> RuntimeError [CP_SHARED_KV_FAIL_FAST][batch_gt1_mla_direct_write_shape_mismatch]
+```
+
+### 3.4 W3 测试
+
+#### T1：batch local loc 保持 request boundary
+
+构造：
+
+```text
+page_size=4, cp_size=2, cp_rank=1
+plan:
+  req0 split [4,0,0,0], zigzag [1,2] -> local []
+  req1 split [4,4,1,0], zigzag [1,2] -> local req1 segment1 + segment2
+```
+
+或选择 cp_rank=0，构造 req0/req1 都有 local rows，便于断言：
+
+```text
+local_out_cache_loc == local(req0) + local(req1)
+```
+
+#### T2：batch owner mismatch 拒绝 direct write
+
+synthetic `out_cache_loc` 中让某个 local page 的 `(page_id - 1) % cp_size` 不等于 `cp_rank`。
+
+期望：
+
+```text
+get_cp_shared_kv_local_out_cache_loc() is None
+warning reason local_loc_owner_mismatch
+```
+
+如果实现采用 fail-fast，则断言对应 fail-fast reason。
+
+#### T3：batch physical loc 缓存
+
+调用两次 `get_cp_shared_kv_local_physical_out_cache_loc()`，断言返回同一对象，并且 loc 映射符合 `CpSharedKVLayout.logical_locs_to_physical()`。
+
+#### T4：index direct write 不 silent fallback
+
+mock `_store_index_k_cache()`，在 bs>1 local loc 不可用时确保不会调用 legacy fallback。
+
+#### T5：MLA direct write shape mismatch fail-fast
+
+构造 bs>1 local loc tokens=N，但 `k_nope/k_pe` tokens != N，断言 fail-fast。
+
+## 4. W4 设计：target index/top-k sync correctness
+
+### 4.1 输入合同
+
+W4 依赖：
+
+- W1 batch metadata；
+- W3 local q/weights 已按 request boundary split；
+- `metadata.get_page_table_64()` 可返回 batch page table；
+- 不启用 bs>1 current reuse / partial-current reuse。
+
+### 4.2 输出合同
+
+`_get_topk_in_seq_cp_pair()` 在 bs>1 下返回：
+
+```text
+concat(
+  topk(req0_prev),
+  topk(req0_next),
+  topk(req1_prev),
+  topk(req1_next),
+  ...
+)
+```
+
+这个顺序必须匹配 local q/weights 顺序，也就是 `split_tensor_by_cp_batch_plan()` 的输出顺序。
+
+### 4.3 实现步骤
+
+#### W4-S1：给 `_get_topk_in_seq_cp_pair()` 增加 batch dispatch
+
+```python
+if getattr(metadata, "batch_size", 1) > 1:
+    return self._get_topk_in_seq_cp_pair_batch(...)
+```
+
+保留原 scalar path。
+
+#### W4-S2：新增 `_get_topk_in_seq_cp_pair_batch()`
+
+伪代码：
+
+```python
+plan = get_cp_shared_kv_batch_plan(forward_batch)
+if current_index_kv is not None:
+    fail-fast batch_gt1_index_current_reuse_unsupported
+
+shared_block_tables = metadata.get_page_table_64()
+shared_index_buffer, shared_block_tables = self._maybe_materialize_shared_index_buffer(
+    forward_batch, layer_id, shared_block_tables
+)
+
+cursor = 0
+outputs = []
+for req_id in range(plan.batch_size):
+    prev_len = metadata.request_actual_seq_q_prev[req_id]
+    next_len = metadata.request_actual_seq_q_next[req_id]
+
+    q_prev = q_fp8[cursor: cursor + prev_len]
+    w_prev = weights[cursor: cursor + prev_len]
+    cursor += prev_len
+
+    q_next = q_fp8[cursor: cursor + next_len]
+    w_next = weights[cursor: cursor + next_len]
+    cursor += next_len
+
+    outputs.append(call_segment(req_id, prev, prev_len, kv_len_prev))
+    outputs.append(call_segment(req_id, next, next_len, kv_len_next))
+
+assert cursor == q_fp8.shape[0]
+return cat(outputs)
+```
+
+zero segment：
+
+```text
+segment_len == 0 -> 返回 shape (0, index_topk) 的 empty tensor，不调用底层 MQA/topk
+```
+
+#### W4-S3：泛化 `_get_topk_ragged_with_cp(..., batch_idx=0)`
+
+新增参数：
+
+```python
+batch_idx: int = 0
+```
+
+并把非 `cp_index` 分支里的 hard-coded `[0]` 改成 `[batch_idx]`：
+
+- `forward_batch.seq_lens_cpu[batch_idx]`
+- `forward_batch.extend_seq_lens_cpu[batch_idx]`
+- `block_tables[batch_idx]`
+
+scalar path 默认 `batch_idx=0`，行为不变。
+
+#### W4-S4：单 segment cu tensor
+
+每个 segment 调用 `_get_topk_ragged_with_cp()` 时传：
+
+```python
+actual_seq_q_tensor = torch.tensor([segment_len], device=q_fp8.device, dtype=torch.int32)
+actual_seq_q_cu_tensor = torch.tensor([0, segment_len], device=q_fp8.device, dtype=torch.int32)
+```
+
+不要传 `request_actual_seq_q_prev_cu_tensor` 的 batch cumulative view。
+
+#### W4-S5：不要使用 `cp_index` 分支
+
+现有 `cp_index` 分支虽然看起来能表达 multi-batch，但源码已标注 accuracy issue。W4 第一版只追 correctness，明确不使用。
+
+### 4.4 W4 测试
+
+#### T1：bs>1 per-request prev/next 调用顺序
+
+mock `_get_topk_ragged_with_cp()`，记录：
+
+- `batch_idx`
+- `kv_len`
+- `actual_seq_q`
+- q slice 内容
+- weights slice 内容
+- `actual_seq_q_cu_tensor`
+
+断言调用顺序：
+
+```text
+req0 prev
+req0 next
+req1 prev
+req1 next
+```
+
+#### T2：materialize 只做一次
+
+mock `_maybe_materialize_shared_index_buffer()`，断言 bs>1 top-k 只调用一次 materialize，并且所有 segment 复用同一个 `shared_index_buffer/shared_block_tables`。
+
+#### T3：zero segment 不调用 top-k kernel
+
+构造某个 request 的 `actual_seq_q_next=0`，断言 `_get_topk_ragged_with_cp()` 不被调用，输出中该 segment 是 empty。
+
+#### T4：bs>1 current_index_kv fail-fast
+
+传入 `current_index_kv`，断言：
+
+```text
+[CP_SHARED_KV_FAIL_FAST][batch_gt1_index_current_reuse_unsupported]
+```
+
+#### T5：scalar tests 保持
+
+现有：
+
+- `test_indexer_in_seq_cp_pair_materializes_index_once_for_prev_next`
+- `test_indexer_in_seq_cp_pair_skips_materialize_when_current_index_reused`
+
+必须继续通过。
+
+## 5. 执行顺序
+
+建议按以下顺序实施，避免把 W3/W4 的 bug 混在一起：
+
+1. W3 tests：batch local loc / owner mismatch / physical loc cache。
+2. W3 implementation：修 `get_cp_shared_kv_local_out_cache_loc()` 的 bs>1 expected length。
+3. W3 direct-write guard：index/MLA bs>1 local loc unavailable 不 silent fallback。
+4. W4 tests：batch top-k per-request order / materialize once / zero segment / current reuse fail-fast。
+5. W4 implementation：batch dispatch + `_get_topk_ragged_with_cp(batch_idx=...)`。
+6. 只跑 unit，不跑 ETE；W2 allocator 未合入前，ETE 仍可能被 allocator blocker 阻塞。
+
+## 6. 验证命令
+
+本地：
+
+```bash
+python -m py_compile \
+  python/sglang/srt/layers/attention/nsa/utils.py \
+  python/sglang/srt/layers/attention/nsa/nsa_indexer.py \
+  python/sglang/srt/models/deepseek_common/attention_forward_methods/forward_mla.py \
+  test/registered/unit/layers/test_nsa_cp_utils.py
+```
+
+远端容器：
+
+```bash
+cd /sgl-workspace/sglang-tai
+PYTHONPATH=python python -m pytest -q \
+  test/registered/unit/layers/test_nsa_cp_utils.py
+```
+
+必要时追加：
+
+```bash
+PYTHONPATH=python python -m pytest -q \
+  test/registered/unit/mem_cache/test_cp_shared_kv_runtime.py
+```
+
+## 7. 风险和后续
+
+### R1. W2 allocator 输出可能是 padded-token loc
+
+W3 当前应按 valid-token loc 设计。如果 W2 返回 padded-token loc，direct write 不能直接消费，否则会把 padding tail 写入真实 KV/index。
+
+需要在 W2/W3 集成时确认：
+
+```text
+out_cache_loc.numel() == sum(request_extend_lens)
+```
+
+或者 W2 提供 valid loc view。
+
+### R2. per-segment top-k 会增加调用次数
+
+W4 第一版 bs=N 会最多调用 `2N` 次 `_get_topk_ragged_with_cp()`。这是 correctness-first 设计。后续性能优化可以引入 batched segment descriptor 或修复 `cp_index` 分支，但不能在第一版混入。
+
+### R3. current reuse 不支持会影响 cache-hit 性能
+
+W4 只保证 sync correctness。cache-hit 的性能收益需要 W5 current/partial-current reuse 恢复后再评估。
+
+### R4. fallback 策略需要保持醒目
+
+W3/W4 支持范围内不能 silent fallback。尤其 index direct-write 当前 fallback 很隐蔽，必须在 bs>1 CP shared-KV 下收窄。
+
+### R5. page descriptor 必须 layer-invariant
+
+如果实现中发现某个 page/table/remap descriptor 每层重建，应记录为性能风险并优先改成 ForwardBatch 缓存。
diff --git a/python/sglang/srt/layers/attention/nsa/nsa_indexer.py b/python/sglang/srt/layers/attention/nsa/nsa_indexer.py
index aabd191d2..50721e7ca 100644
--- a/python/sglang/srt/layers/attention/nsa/nsa_indexer.py
+++ b/python/sglang/srt/layers/attention/nsa/nsa_indexer.py
@@ -67,8 +67,8 @@ from sglang.srt.layers.attention.nsa.utils import (
     get_cp_shared_kv_local_physical_out_cache_loc,
     is_nsa_enable_prefill_cp,
     is_nsa_prefill_cp_in_seq_split,
-    log_cp_shared_kv_direct_write_fallback,
     nsa_use_prefill_cp,
+    raise_cp_shared_kv_direct_write_error,
     split_in_seq_cp_local_pair,
 )
 from sglang.srt.layers.communicator import ScatterMode
@@ -1618,7 +1618,7 @@ class Indexer(MultiPlatformOp):
         if local_out_loc is None:
             return False
         if local_key.shape[0] != local_out_loc.numel():
-            log_cp_shared_kv_direct_write_fallback(
+            raise_cp_shared_kv_direct_write_error(
                 "index_local_shape_mismatch",
                 "NSA index local key token count does not match local out_cache_loc: "
                 "local_key=%s local_out_cache_loc=%s layer_id=%s",
@@ -1626,7 +1626,6 @@ class Indexer(MultiPlatformOp):
                 local_out_loc.numel(),
                 layer_id,
             )
-            return False
         if local_out_loc.numel() == 0:
             return True
 
diff --git a/python/sglang/srt/layers/attention/nsa/utils.py b/python/sglang/srt/layers/attention/nsa/utils.py
index 241072f33..6cae1459d 100644
--- a/python/sglang/srt/layers/attention/nsa/utils.py
+++ b/python/sglang/srt/layers/attention/nsa/utils.py
@@ -65,6 +65,21 @@ def log_cp_shared_kv_direct_write_fallback(
     )
 
 
+def raise_cp_shared_kv_direct_write_error(
+    reason: str,
+    message: str,
+    *args,
+) -> None:
+    """Fail fast when CP shared-KV direct-write invariants are violated."""
+
+    formatted = message % args if args else message
+    error_msg = (
+        f"[CP_SHARED_KV_FAIL_FAST][direct_write] reason={reason} {formatted}"
+    )
+    logger.error(error_msg)
+    raise RuntimeError(error_msg)
+
+
 def compute_nsa_seqlens(original_seq_lens, nsa_index_topk: int):
     return original_seq_lens.clamp(max=nsa_index_topk)
 
@@ -1055,51 +1070,51 @@ def get_cp_shared_kv_local_out_cache_loc(forward_batch: "ForwardBatch"):
     layout = getattr(forward_batch, "cp_shared_kv_layout", None)
     out_cache_loc = getattr(forward_batch, "out_cache_loc", None)
     if metadata is None:
-        log_cp_shared_kv_direct_write_fallback(
+        raise_cp_shared_kv_direct_write_error(
             "missing_metadata",
             "nsa_cp_metadata is missing",
         )
-        return None
     if layout is None:
-        log_cp_shared_kv_direct_write_fallback(
+        raise_cp_shared_kv_direct_write_error(
             "missing_layout",
             "cp_shared_kv_layout is missing",
         )
-        return None
     if out_cache_loc is None:
-        log_cp_shared_kv_direct_write_fallback(
+        raise_cp_shared_kv_direct_write_error(
             "missing_out_cache_loc",
             "out_cache_loc is missing",
         )
-        return None
     if not getattr(metadata, "page_aligned", False):
-        log_cp_shared_kv_direct_write_fallback(
+        raise_cp_shared_kv_direct_write_error(
             "not_page_aligned",
             "metadata is not page-aligned: page_size=%s extend_prefix_len=%s",
             getattr(metadata, "page_size", None),
             getattr(metadata, "extend_prefix_len", None),
         )
-        return None
     try:
         in_seq_split = is_nsa_prefill_cp_in_seq_split()
     except ValueError:
         in_seq_split = True
     if not in_seq_split:
-        log_cp_shared_kv_direct_write_fallback(
+        raise_cp_shared_kv_direct_write_error(
             "not_in_seq_split",
             "nsa_prefill_cp_mode is not in-seq-split",
         )
-        return None
-    split_tokens = sum(int(x) for x in metadata.split_list)
+    batch_plan = get_cp_shared_kv_batch_plan(forward_batch)
+    if batch_plan is not None and int(getattr(batch_plan, "batch_size", 1) or 1) > 1:
+        split_tokens = sum(int(x) for x in getattr(batch_plan, "request_extend_lens", []))
+        mismatch_reason = "batch_split_out_cache_len_mismatch"
+    else:
+        split_tokens = sum(int(x) for x in metadata.split_list)
+        mismatch_reason = "split_out_cache_len_mismatch"
     out_cache_tokens = int(out_cache_loc.numel())
     if split_tokens != out_cache_tokens:
-        log_cp_shared_kv_direct_write_fallback(
-            "split_out_cache_len_mismatch",
+        raise_cp_shared_kv_direct_write_error(
+            mismatch_reason,
             "split_list tokens=%s out_cache_loc tokens=%s",
             split_tokens,
             out_cache_tokens,
         )
-        return None
 
     local_out_cache_loc = cp_split_and_rebuild_1d(
         forward_batch,
@@ -1111,14 +1126,13 @@ def get_cp_shared_kv_local_out_cache_loc(forward_batch: "ForwardBatch"):
 
     valid_locs = local_out_cache_loc[local_out_cache_loc > 0]
     if valid_locs.numel() > 0 and not torch.all(layout.owned_by_this_rank(valid_locs)):
-        log_cp_shared_kv_direct_write_fallback(
+        raise_cp_shared_kv_direct_write_error(
             "local_loc_owner_mismatch",
             "local out_cache_loc contains pages not owned by this rank: cp_rank=%s cp_size=%s page_size=%s",
             layout.cp_rank,
             layout.cp_size,
             layout.page_size,
         )
-        return None
 
     forward_batch.cp_local_out_cache_loc = local_out_cache_loc
     return local_out_cache_loc
diff --git a/python/sglang/srt/models/deepseek_common/attention_forward_methods/forward_mla.py b/python/sglang/srt/models/deepseek_common/attention_forward_methods/forward_mla.py
index 9ba806756..6198523a3 100644
--- a/python/sglang/srt/models/deepseek_common/attention_forward_methods/forward_mla.py
+++ b/python/sglang/srt/models/deepseek_common/attention_forward_methods/forward_mla.py
@@ -10,8 +10,8 @@ from sglang.srt.layers.attention.nsa.utils import (
     get_cp_shared_kv_local_out_cache_loc,
     get_cp_shared_kv_local_physical_out_cache_loc,
     log_cp_draft_shared_kv_debug,
-    log_cp_shared_kv_direct_write_fallback,
     nsa_use_prefill_cp,
+    raise_cp_shared_kv_direct_write_error,
 )
 from sglang.srt.layers.attention.nsa.cp_shared_kv_runtime import (
     should_reuse_current_extend_kv,
@@ -561,7 +561,7 @@ class DeepseekMLAForwardMixin:
             k_nope.shape[0] != local_out_cache_loc.numel()
             or k_pe.shape[0] != local_out_cache_loc.numel()
         ):
-            log_cp_shared_kv_direct_write_fallback(
+            raise_cp_shared_kv_direct_write_error(
                 "mla_local_shape_mismatch",
                 "MLA local KV token count does not match local out_cache_loc: "
                 "k_nope=%s k_pe=%s local_out_cache_loc=%s layer_id=%s",
@@ -570,7 +570,6 @@ class DeepseekMLAForwardMixin:
                 local_out_cache_loc.numel(),
                 self.attn_mqa.layer_id,
             )
-            return False
         if local_out_cache_loc.numel() == 0:
             return True
 
diff --git a/test/registered/unit/layers/test_nsa_cp_utils.py b/test/registered/unit/layers/test_nsa_cp_utils.py
index e39e7353b..e7e2c5e6d 100644
--- a/test/registered/unit/layers/test_nsa_cp_utils.py
+++ b/test/registered/unit/layers/test_nsa_cp_utils.py
@@ -709,6 +709,95 @@ class TestNSAInSeqCPUtils(unittest.TestCase):
             + list(range(6 * page_size, 7 * page_size)),
         )
 
+    def test_batch_local_out_cache_loc_keeps_request_boundaries(self):
+        import torch
+        from types import SimpleNamespace
+
+        page_size = 4
+        # cp_size=2/cp_rank=1 selects segment 1 and 2 for each request.
+        # req0 has no rank-1 local rows. req1 contributes segment 1 (page 2)
+        # and segment 2 (tail page 4).  The synthetic logical page ids encode
+        # the owner-lane invariant through (page_id - 1) % cp_size.
+        out_cache_loc = torch.cat(
+            [
+                torch.arange(1 * page_size, 2 * page_size),  # req0 seg0 owner 0
+                torch.arange(3 * page_size, 4 * page_size),  # req1 seg0 owner 0
+                torch.arange(2 * page_size, 3 * page_size),  # req1 seg1 owner 1
+                torch.tensor([4 * page_size]),  # req1 seg2 owner 1 tail
+            ]
+        )
+        forward_batch = SimpleNamespace(
+            uses_cp_shared_kv=True,
+            cp_shared_kv_layout=CpSharedKVLayout(
+                page_size=page_size,
+                cp_size=2,
+                cp_rank=1,
+            ),
+            nsa_cp_metadata=NSAContextParallelMetadata(
+                batch_size=2,
+                split_list=[4, 0, 0, 0],
+                zigzag_index=[1, 2],
+                page_aligned=True,
+                page_size=page_size,
+                extend_prefix_len=0,
+                request_extend_lens=[4, 9],
+                request_split_lists=[[4, 0, 0, 0], [4, 4, 1, 0]],
+                request_zigzag_indices=[[1, 2], [1, 2]],
+            ),
+            out_cache_loc=out_cache_loc,
+        )
+
+        local_locs = get_cp_shared_kv_local_out_cache_loc(forward_batch)
+
+        self.assertIsNotNone(local_locs)
+        self.assertEqual(
+            local_locs.tolist(),
+            list(range(2 * page_size, 3 * page_size)) + [4 * page_size],
+        )
+
+    def test_batch_local_physical_out_cache_loc_reuses_layer_invariant_plan(self):
+        import torch
+        from types import SimpleNamespace
+
+        page_size = 4
+        out_cache_loc = torch.cat(
+            [
+                torch.arange(1 * page_size, 2 * page_size),
+                torch.arange(3 * page_size, 4 * page_size),
+                torch.arange(2 * page_size, 3 * page_size),
+                torch.tensor([4 * page_size]),
+            ]
+        )
+        forward_batch = SimpleNamespace(
+            uses_cp_shared_kv=True,
+            cp_shared_kv_layout=CpSharedKVLayout(
+                page_size=page_size,
+                cp_size=2,
+                cp_rank=1,
+            ),
+            nsa_cp_metadata=NSAContextParallelMetadata(
+                batch_size=2,
+                split_list=[4, 0, 0, 0],
+                zigzag_index=[1, 2],
+                page_aligned=True,
+                page_size=page_size,
+                extend_prefix_len=0,
+                request_extend_lens=[4, 9],
+                request_split_lists=[[4, 0, 0, 0], [4, 4, 1, 0]],
+                request_zigzag_indices=[[1, 2], [1, 2]],
+            ),
+            out_cache_loc=out_cache_loc,
+        )
+
+        physical_locs = get_cp_shared_kv_local_physical_out_cache_loc(forward_batch)
+        second_read = get_cp_shared_kv_local_physical_out_cache_loc(forward_batch)
+
+        self.assertIs(physical_locs, second_read)
+        self.assertEqual(
+            physical_locs.tolist(),
+            list(range(1 * page_size, 2 * page_size)) + [2 * page_size],
+        )
+
     def test_local_physical_out_cache_loc_is_cached(self):
         import torch
         from types import SimpleNamespace
@@ -748,7 +837,7 @@ class TestNSAInSeqCPUtils(unittest.TestCase):
             + list(range(2 * page_size, 3 * page_size)),
         )
 
-    def test_local_out_cache_loc_falls_back_when_owner_mismatch(self):
+    def test_local_out_cache_loc_fails_fast_when_owner_mismatch(self):
         import torch
         from types import SimpleNamespace
 
@@ -771,14 +860,16 @@ class TestNSAInSeqCPUtils(unittest.TestCase):
             out_cache_loc=out_cache_loc,
         )
 
-        self.assertIsNone(get_cp_shared_kv_local_out_cache_loc(forward_batch))
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "CP_SHARED_KV_FAIL_FAST.*local_loc_owner_mismatch",
+        ):
+            get_cp_shared_kv_local_out_cache_loc(forward_batch)
 
-    def test_local_out_cache_loc_logs_every_fallback_event(self):
+    def test_local_out_cache_loc_fails_fast_every_invalid_event(self):
         import torch
         from types import SimpleNamespace
 
-        from sglang.srt.layers.attention.nsa import utils as nsa_utils
-
         page_size = 4
         forward_batch = SimpleNamespace(
             uses_cp_shared_kv=True,
@@ -798,17 +889,81 @@ class TestNSAInSeqCPUtils(unittest.TestCase):
         )
 
         with self.assertLogs(
-            "sglang.srt.layers.attention.nsa.utils", level="WARNING"
+            "sglang.srt.layers.attention.nsa.utils", level="ERROR"
         ) as cm:
-            self.assertIsNone(get_cp_shared_kv_local_out_cache_loc(forward_batch))
-            self.assertIsNone(get_cp_shared_kv_local_out_cache_loc(forward_batch))
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "CP_SHARED_KV_FAIL_FAST.*not_page_aligned",
+            ):
+                get_cp_shared_kv_local_out_cache_loc(forward_batch)
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "CP_SHARED_KV_FAIL_FAST.*not_page_aligned",
+            ):
+                get_cp_shared_kv_local_out_cache_loc(forward_batch)
 
         self.assertEqual(len(cm.output), 2)
-        self.assertIn("[CP_SHARED_KV_FALLBACK][direct_write]", cm.output[0])
+        self.assertIn("[CP_SHARED_KV_FAIL_FAST][direct_write]", cm.output[0])
         self.assertIn("metadata is not page-aligned", cm.output[0])
-        self.assertIn("[CP_SHARED_KV_FALLBACK][direct_write]", cm.output[1])
+        self.assertIn("[CP_SHARED_KV_FAIL_FAST][direct_write]", cm.output[1])
         self.assertIn("metadata is not page-aligned", cm.output[1])
 
+    def test_indexer_direct_write_fails_fast_on_local_shape_mismatch(self):
+        import torch
+
+        from sglang.srt.layers.attention.nsa import nsa_indexer
+        from sglang.srt.layers.attention.nsa.nsa_indexer import Indexer
+
+        indexer = object.__new__(Indexer)
+        indexer.nsa_enable_prefill_cp = True
+        forward_batch = SimpleNamespace(nsa_cp_metadata=object())
+
+        with (
+            patch.object(nsa_indexer, "nsa_use_prefill_cp", return_value=True),
+            patch.object(
+                nsa_indexer,
+                "get_cp_shared_kv_local_out_cache_loc",
+                return_value=torch.tensor([1, 2], dtype=torch.int64),
+            ),
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "CP_SHARED_KV_FAIL_FAST.*index_local_shape_mismatch",
+            ):
+                Indexer._store_cp_shared_local_index_k_cache(
+                    indexer,
+                    forward_batch,
+                    layer_id=0,
+                    local_key=torch.empty((1, 8)),
+                    act_quant=None,
+                )
+
+    def test_mla_direct_write_fails_fast_on_local_shape_mismatch(self):
+        import torch
+
+        from sglang.srt.models.deepseek_common.attention_forward_methods import (
+            forward_mla,
+        )
+
+        mla = SimpleNamespace(attn_mqa=SimpleNamespace(layer_id=3))
+        forward_batch = SimpleNamespace()
+
+        with patch.object(
+            forward_mla,
+            "get_cp_shared_kv_local_out_cache_loc",
+            return_value=torch.tensor([1, 2], dtype=torch.int64),
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "CP_SHARED_KV_FAIL_FAST.*mla_local_shape_mismatch",
+            ):
+                forward_mla.DeepseekMLAForwardMixin._maybe_write_cp_shared_local_mla_kv(
+                    mla,
+                    forward_batch,
+                    k_nope=torch.empty((1, 8)),
+                    k_pe=torch.empty((2, 8)),
+                )
+
     def test_indexer_direct_write_does_not_log_missing_metadata_for_non_cp_batch(self):
         import torch
         from types import SimpleNamespace