Keep CP HiCache draft KV invisible until joint readiness

CP HiCache now treats draft KV as a strict target-owned payload through pending write visibility, host eviction, and state-buffer registration. Host metadata created before async D2H ack is no longer request-visible, so match_prefix cannot select an in-flight host node. Draft host eviction now fails before target cleanup when draft metadata is missing, and prefill/decode share one helper for draft NSA state buffers so shared-KV mode cannot silently skip mismatched draft state.

Constraint: CP shared KV + HiCache + EAGLE/MTP must not expose target-only host hits or skipped draft state as valid cache hits

Rejected: Rely on event-loop ordering and lock_ref to hide in-flight writes | match_prefix does not consult lock_ref and can observe host_len/cp_hicache directly

Rejected: Keep draft state mismatch as debug-only skip | it can poison speculative acceptance while looking like a successful cache hit

Confidence: high

Scope-risk: moderate

Directive: Do not reintroduce silent draft/target fallback in CP shared-KV HiCache paths; malformed strong-sync metadata should fail fast

Tested: python -m py_compile targeted modified files

Tested: remote g0034 container pytest test/registered/unit/mem_cache/test_cp_hicache_metadata.py test/registered/unit/managers/test_hicache_controller_cp.py test/registered/unit/disaggregation/test_cp_shared_kv_transfer_mapping.py -q (91 passed)

Not-tested: Full CP shared KV + HiCache + EAGLE/MTP ETE server run after this commit

test/registered/unit/disaggregation/test_cp_shared_kv_transfer_mapping.py

@@ -1,8 +1,11 @@
 import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
 
 import numpy as np
 
 from sglang.srt.disaggregation.utils import (
+    append_cp_draft_state_buffers,
     filter_kv_pages_for_cp_shared_kv,
     select_pages_by_request_positions,
 )
@@ -56,6 +59,91 @@ class TestCPSharedKVTransferMapping(unittest.TestCase):
         self.assertEqual(logical_positions.tolist(), [2, 6])
         self.assertEqual(selected_decode_pages.tolist(), [43, 47])
 
+    def test_append_cp_draft_state_buffers_registers_nsa_state_as_target_payload(self):
+        kv_args = SimpleNamespace(
+            state_type="nsa",
+            state_data_ptrs=[11],
+            state_data_lens=[12],
+            state_item_lens=[13],
+        )
+
+        appended = append_cp_draft_state_buffers(
+            kv_args,
+            draft_state_type="nsa",
+            draft_state_data_ptrs=[21, 22],
+            draft_state_data_lens=[23, 24],
+            draft_state_item_lens=[25, 26],
+            role="prefill",
+            cp_rank=2,
+        )
+
+        self.assertTrue(appended)
+        self.assertEqual(kv_args.draft_state_type, "nsa")
+        self.assertEqual(kv_args.draft_state_buffer_start, 1)
+        self.assertEqual(kv_args.draft_state_buffer_count, 2)
+        self.assertEqual(kv_args.state_data_ptrs, [11, 21, 22])
+        self.assertEqual(kv_args.state_data_lens, [12, 23, 24])
+        self.assertEqual(kv_args.state_item_lens, [13, 25, 26])
+
+    def test_append_cp_draft_state_buffers_rejects_state_mismatch_under_shared_kv(self):
+        kv_args = SimpleNamespace(
+            state_type="nsa",
+            state_data_ptrs=[11],
+            state_data_lens=[12],
+            state_item_lens=[13],
+        )
+
+        with patch(
+            "sglang.srt.disaggregation.utils.envs.SGLANG_CP_DRAFT_SHARED_KV.get",
+            return_value=True,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "prefill.*cp_rank=2.*target_state_type=nsa.*draft_state_type=mamba",
+            ):
+                append_cp_draft_state_buffers(
+                    kv_args,
+                    draft_state_type="mamba",
+                    draft_state_data_ptrs=[21],
+                    draft_state_data_lens=[22],
+                    draft_state_item_lens=[23],
+                    role="prefill",
+                    cp_rank=2,
+                )
+
+        self.assertFalse(hasattr(kv_args, "draft_state_buffer_count"))
+        self.assertEqual(kv_args.state_data_ptrs, [11])
+
+    def test_append_cp_draft_state_buffers_keeps_legacy_skip_when_shared_kv_disabled(
+        self,
+    ):
+        kv_args = SimpleNamespace(
+            state_type="nsa",
+            state_data_ptrs=[11],
+            state_data_lens=[12],
+            state_item_lens=[13],
+        )
+
+        with patch(
+            "sglang.srt.disaggregation.utils.envs.SGLANG_CP_DRAFT_SHARED_KV.get",
+            return_value=False,
+        ):
+            appended = append_cp_draft_state_buffers(
+                kv_args,
+                draft_state_type="mamba",
+                draft_state_data_ptrs=[21],
+                draft_state_data_lens=[22],
+                draft_state_item_lens=[23],
+                role="decode",
+                cp_rank=3,
+            )
+
+        self.assertFalse(appended)
+        self.assertEqual(kv_args.draft_state_type, "mamba")
+        self.assertEqual(kv_args.draft_state_buffer_start, 1)
+        self.assertEqual(kv_args.draft_state_buffer_count, 0)
+        self.assertEqual(kv_args.state_data_ptrs, [11])
+
 
 if __name__ == "__main__":
     unittest.main()