Fix stale batch fullness after chunked prefill

python/sglang/srt/managers/scheduler.py

@@ -3939,13 +3939,8 @@ def _get_new_batch_prefill_raw(
             for req in ready_grammar_requests:
                 self._add_request_to_queue(req)
 
-        if self.enable_priority_preemption or self.is_hybrid_swa:
-            # Reset batch_is_full to try preemption with a prefill adder.
-            running_batch.batch_is_full = False
-        if running_batch.is_empty():
-            # An empty batch never reaches update_running_batch, the only
-            # other place that clears the flag, so admission would stay shut.
-            running_batch.batch_is_full = False
+        # Reset batch_is_full to retry admission with a prefill adder.
+        running_batch.batch_is_full = False
 
         if (
             running_batch.batch_is_full or len(self.waiting_queue) == 0

test/registered/unit/managers/test_scheduler_chunked_req_gate.py

@@ -3,7 +3,7 @@
 import unittest
 from array import array
 from types import SimpleNamespace
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 from sglang.test.ci.ci_register import register_cpu_ci
 from sglang.test.test_utils import CustomTestCase, maybe_stub_sgl_kernel
@@ -155,5 +155,37 @@ def test_no_chunked_req_never_mutates_state(self):
         cache.checkpoint.assert_not_called()
 
 
+class TestChunkedPrefillBatchFullReset(CustomTestCase):
+    def test_stale_batch_full_reconsiders_waiting_request(self):
+        scheduler = Scheduler.__new__(Scheduler)
+        scheduler.grammar_manager = MagicMock()
+        scheduler.grammar_manager.has_waiting_grammars.return_value = False
+        scheduler.enable_hierarchical_cache = False
+        scheduler.enable_priority_preemption = False
+        scheduler.is_hybrid_swa = False
+        scheduler.waiting_queue = [MagicMock()]
+        scheduler.chunked_req = None
+        scheduler.min_free_slots_delayer = None
+        scheduler.processed_tokens_counter = 0
+        scheduler.get_num_allocatable_reqs = MagicMock(return_value=1)
+        scheduler.policy = MagicMock()
+        scheduler.policy.calc_priority.side_effect = RuntimeError("admission attempted")
+
+        running_batch = MagicMock()
+        running_batch.batch_is_full = True
+        running_batch.is_empty.return_value = False
+        running_batch.reqs = [MagicMock(), MagicMock(), MagicMock()]
+
+        with patch(
+            "sglang.srt.managers.scheduler.get_memory",
+            return_value=SimpleNamespace(enable_flexkv=False),
+        ), self.assertRaisesRegex(RuntimeError, "admission attempted"):
+            Scheduler._get_new_batch_prefill_raw(
+                scheduler,
+                prefill_delayer_single_pass=None,
+                running_batch=running_batch,
+            )
+
+
 if __name__ == "__main__":
     unittest.main()