Account decode handoff queues in DP load snapshots

Decode DP dispatch was collapsing onto a few ranks because the controller only randomizes among exact minimum load pairs. The load snapshot undercounted decode handoff work: pending prefill-info requests were absent, and DecodeRequest wrappers in prealloc/transfer queues were skipped because their rid lives on .req.

This makes scheduler load accounting unwrap DecodeRequest items and include pending decode requests, so TOTAL_TOKENS sees queued handoff backlog instead of repeatedly treating busy ranks as empty.

Constraint: Do not mask imbalance with synthetic per-request token penalties; dispatch should be driven by accurate observed load.

Rejected: Add req*4000 or other queue penalties | heuristic, workload-dependent, and hides the accounting bug.

Confidence: medium

Scope-risk: moderate

Directive: Any new decode handoff queue must be included in get_load() or DP routing can regress to stale/min-load collapse.

Tested: Remote cjy-glm5-new: PYTHONPATH=python python -m pytest -q test/registered/unit/observability/test_scheduler_metrics_load.py test/registered/unit/managers/test_prefill_adder.py -> 27 passed.

Not-tested: Fresh decode ETE distribution after service restart.

python/sglang/srt/observability/scheduler_metrics_mixin.py

@@ -679,7 +679,33 @@ class SchedulerMetricsMixin:
                     / self.stats.max_running_requests_under_SLO,
                     self.stats.token_usage / 0.9,
                 )
+    def _unwrap_load_req(self, item):
+        # Decode disaggregation queues store DecodeRequest wrappers, while
+        # normal waiting/retracted queues store Req directly.  Load accounting
+        # must see the underlying request in both cases; otherwise DP dispatch
+        # loses queued decode work from reqs_list and can repeatedly choose the
+        # same few ranks under bursty PD traffic.
+        return getattr(item, "req", item)
+
+    def _get_load_item_num_tokens(self, item):
+        num_tokens = getattr(item, "seqlen", None)
+        if num_tokens is not None:
+            return num_tokens
+
+        req = self._unwrap_load_req(item)
+        num_tokens = getattr(req, "seqlen", None)
+        if num_tokens is not None:
+            return num_tokens
+
+        origin_input_ids = getattr(req, "origin_input_ids", None)
+        output_ids = getattr(req, "output_ids", None)
+        prompt_tokens = len(origin_input_ids) if origin_input_ids is not None else 0
+        completion_tokens = len(output_ids) if output_ids is not None else 0
+        return prompt_tokens + completion_tokens
+
     def _build_request_info(self, req):
+        req = self._unwrap_load_req(req)
+
         prompt_tokens = 0
         if getattr(req, "origin_input_ids_unpadded", None) is not None:
             prompt_tokens = len(req.origin_input_ids_unpadded)
@@ -703,16 +729,17 @@ class SchedulerMetricsMixin:
             completion_tokens=completion_tokens,
             timestamp=time.perf_counter(),
         )
-    def _collect_request_infos(self, queues,request_infos):
-       
+
+    def _collect_request_infos(self, queues, request_infos):
         # all queues
         for queue in queues:
-            for req in queue:
-                 rid = getattr(req, "rid", None)
-                 if rid is None:
-                    #  logger.info("rid is None, skipping request")
-                     continue
-                 request_infos.append(self._build_request_info(req))
+            for item in queue:
+                req = self._unwrap_load_req(item)
+                rid = getattr(req, "rid", None)
+                if rid is None:
+                    # logger.info("rid is None, skipping request")
+                    continue
+                request_infos.append(self._build_request_info(req))
 
     def get_load(self: Scheduler, _: GetLoadReqInput = None) -> GetLoadReqOutput:
         if self.is_hybrid_swa:
@@ -728,13 +755,22 @@ class SchedulerMetricsMixin:
         if self.disaggregation_mode == DisaggregationMode.PREFILL:
             waiting_queues.append(self.disagg_prefill_bootstrap_queue.queue)
         elif self.disaggregation_mode == DisaggregationMode.DECODE:
+            # Include every decode handoff stage.  pending_reqs are not yet
+            # wrapped in DecodeRequest and were previously invisible to the DP
+            # controller, so burst routing could keep dispatching to a rank with
+            # a large unresolved handoff backlog.
+            waiting_queues.append(self.disagg_decode_prealloc_queue.pending_reqs)
             waiting_queues.append(self.disagg_decode_prealloc_queue.queue)
             waiting_queues.append(self.disagg_decode_transfer_queue.queue)
             waiting_queues.append(self.disagg_decode_prealloc_queue.retracted_queue)
-        num_tokens += sum(req.seqlen for queue in waiting_queues for req in queue)
+        num_tokens += sum(
+            self._get_load_item_num_tokens(item)
+            for queue in waiting_queues
+            for item in queue
+        )
         num_waiting_reqs = sum(len(queue) for queue in waiting_queues)
-        request_infos=[]
-        self._collect_request_infos(waiting_queues,request_infos)
+        request_infos = []
+        self._collect_request_infos(waiting_queues, request_infos)
         # logger.info("current dp rank %d num tokens %d  waiting queues %d",self.dp_rank,num_tokens,num_waiting_reqs)
 
         return GetLoadReqOutput(