deflake priority below-threshold test (#25809)

test/registered/scheduler/test_priority_scheduling.py

@@ -198,21 +198,24 @@ class TestPriorityScheduling(CustomTestCase):
     def test_priority_scheduling_preemption_below_threshold_validation(self):
         """Verify running requests are not preempted by requests with priorities below preemption threshold"""
 
-        responses = asyncio.run(
-            send_concurrent_generate_requests_with_custom_params(
+        # Stagger sends so priority=0 occupies the running queue before
+        # priority=5 arrives -- asyncio.gather gives no arrival-order guarantee.
+        # ignore_eos on both: priority=0 stays running when priority=5 arrives
+        # (exercises the no-preempt path), and priority=5's runtime must exceed
+        # the stagger so its server-side e2e_latency stays > priority=0's.
+        async def _send(priority, **sampling):
+            return await send_concurrent_generate_requests_with_custom_params(
                 self.base_url,
-                [
-                    {
-                        "priority": 0,
-                        "sampling_params": {"max_new_tokens": 10000},
-                    },
-                    {
-                        "priority": 5,
-                        "sampling_params": {"max_new_tokens": 10000},
-                    },
-                ],
+                [{"priority": priority, "sampling_params": sampling}],
             )
-        )
+
+        async def _run():
+            first = asyncio.create_task(_send(0, max_new_tokens=1000, ignore_eos=True))
+            await asyncio.sleep(1.0)
+            second = asyncio.create_task(_send(5, max_new_tokens=1000, ignore_eos=True))
+            return (await first) + (await second)
+
+        responses = asyncio.run(_run())
 
         expected_status_and_error_messages = [
             (200, None),