[DSV4] Fix draft-extend CUDA graph padding and WAR ordering

python/sglang/srt/layers/attention/base_attn_backend.py

@@ -128,10 +128,13 @@ def draft_extend_metadata_captured_in_graph(self) -> bool:
     def shared_read_boundary(self, forward_mode: ForwardMode) -> SharedReadBoundary:
         """Declare where this backend's scheduler-shared reads end per mode.
 
-        Decode/verify default to IN_REPLAY: the out-graph/in-graph init
-        contract above makes it a safe upper bound for any backend honoring
-        the contract. Override for audited deviations.
+        Draft extend defaults to PRE_REPLAY after its out-of-graph metadata
+        initialization. Decode/verify default to IN_REPLAY: the
+        out-graph/in-graph init contract above makes it a safe upper bound for
+        any backend honoring the contract. Override for audited deviations.
         """
+        if forward_mode.is_draft_extend_v2():
+            return SharedReadBoundary.PRE_REPLAY
         if forward_mode.is_decode() or forward_mode.is_target_verify():
             return SharedReadBoundary.IN_REPLAY
         return SharedReadBoundary.UNKNOWN

python/sglang/srt/layers/attention/deepseek_v4_backend.py

@@ -505,8 +505,9 @@ class DeepseekV4AttnBackend(
     needs_cpu_seq_lens: bool = False
 
     def shared_read_boundary(self, forward_mode: ForwardMode) -> SharedReadBoundary:
-        # Breakable-graph verify rereads shared state across segments.
-        if forward_mode.is_target_verify():
+        # Breakable-graph verify rereads shared state across segments. Draft
+        # extend also consumes scheduler-owned buffers during graph replay.
+        if forward_mode.is_target_verify() or forward_mode.is_draft_extend_v2():
             return SharedReadBoundary.POST_REPLAY
         return super().shared_read_boundary(forward_mode)
 
@@ -1969,6 +1970,9 @@ def expand_extend_with_same_length(
         seq_lens_casual = seq_lens[:, None] + torch.arange(
             -qo_len + 1, 1, **self.cuda_int32_kwargs
         )
+        # Graph-padded requests use seq_len=1 even when qo_len is wider. Keep
+        # their causal rows on reserved slot 0 instead of producing negatives.
+        seq_lens_casual.clamp_min_(1)
         seq_lens_casual = seq_lens_casual.flatten()
         idx_to_req_repeated = torch.arange(
             bs, **self.cuda_int32_kwargs

python/sglang/srt/speculative/eagle_draft_extend_cuda_graph_runner.py

@@ -7,6 +7,7 @@
 import torch
 
 from sglang.srt.compilation.torch_compile_decoration import set_torch_compile_config
+from sglang.srt.layers.attention.base_attn_backend import SharedReadBoundary
 from sglang.srt.layers.dp_attention import (
     DpPaddingMode,
     set_dp_buffer_len,
@@ -589,21 +590,29 @@ def execute(self, forward_batch: ForwardBatch):
         )
         self.draft_extend_attn_backend.init_forward_metadata_out_graph(fb_view)
 
-        # Snapshot built -- the forward is done reading the shared pool. Publish
-        # a read-done event the scheduler's WAR barrier waits on (draft extend
-        # is the EAGLE-family war-publish phase; last write wins the mailbox).
-        read_done = self.device_module.Event()
-        read_done.record()
-        self.model_runner.war_fastpath_read_done_event = read_done
-
         self.raw_bs = raw_bs
         self.bs = bs
         shape_key = self._make_graph_key(bs)
         with device_timer_ctx(self.model_runner.device_timer, "eagle_draft_extend"):
-            out = self._replay_graph(shape_key, forward_batch)
+            out = self._replay_graph_with_war_read_done(shape_key, forward_batch)
 
         out = LogitsProcessorOutput(
             next_token_logits=out.next_token_logits[:num_tokens],
             hidden_states=out.hidden_states[:num_tokens],
         )
         return out
+
+    def _replay_graph_with_war_read_done(self, shape_key, forward_batch):
+        read_boundary = self.draft_extend_attn_backend.shared_read_boundary(
+            self.forward_mode
+        )
+        # This runner does not plant an external event in its captured graph,
+        # so an in-replay declaration must conservatively publish pre-replay.
+        if read_boundary is SharedReadBoundary.IN_REPLAY:
+            read_boundary = SharedReadBoundary.PRE_REPLAY
+        if read_boundary is SharedReadBoundary.PRE_REPLAY:
+            self._publish_war_read_done(in_graph=False)
+        out = self._replay_graph(shape_key, forward_batch)
+        if read_boundary is SharedReadBoundary.POST_REPLAY:
+            self._publish_war_read_done(in_graph=False)
+        return out

test/registered/unit/spec/test_dsv4_draft_extend_cuda_graph.py

@@ -0,0 +1,59 @@
+import unittest
+from types import SimpleNamespace
+from unittest import TestCase, mock
+
+import torch
+
+from sglang.srt.layers.attention.base_attn_backend import AttentionBackend
+from sglang.srt.layers.attention.deepseek_v4_backend import DeepseekV4AttnBackend
+from sglang.srt.speculative.eagle_draft_extend_cuda_graph_runner import (
+    EAGLEDraftExtendCudaGraphRunner,
+)
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
+
+class TestDSV4DraftExtendCudaGraph(TestCase):
+    def test_padding_causal_lengths_stay_nonnegative(self):
+        backend = object.__new__(DeepseekV4AttnBackend)
+        backend.cuda_int32_kwargs = {"dtype": torch.int32, "device": "cpu"}
+
+        seq_lens_casual, req_pool_indices = backend.expand_extend_with_same_length(
+            bs=1,
+            qo_len=4,
+            seq_lens=torch.tensor([1], dtype=torch.int32),
+            req_pool_indices=torch.tensor([0], dtype=torch.int32),
+        )
+
+        self.assertEqual(seq_lens_casual.tolist(), [1, 1, 1, 1])
+        self.assertEqual((seq_lens_casual - 1).tolist(), [0, 0, 0, 0])
+        self.assertEqual(req_pool_indices.tolist(), [0, 0, 0, 0])
+
+    def test_read_done_order_follows_backend_capability(self):
+        cases = (
+            (AttentionBackend, ["event", "replay"]),
+            (DeepseekV4AttnBackend, ["replay", "event"]),
+        )
+        for backend, expected_order in cases:
+            with self.subTest(backend=backend.__name__):
+                runner = object.__new__(EAGLEDraftExtendCudaGraphRunner)
+                runner.draft_extend_attn_backend = backend
+                call_order = []
+                runner._record_war_fastpath_read_done = mock.Mock(
+                    side_effect=lambda: call_order.append("event")
+                )
+                runner._replay_graph = mock.Mock(
+                    side_effect=lambda *_args: call_order.append("replay") or "output"
+                )
+
+                output = runner._replay_graph_with_war_read_done(
+                    SimpleNamespace(), SimpleNamespace()
+                )
+
+                self.assertEqual(output, "output")
+                self.assertEqual(call_order, expected_order)
+
+
+if __name__ == "__main__":
+    unittest.main()