Expose predicted per-Instance queue pressure and use it in least_load scoring

proxy/README.md

@@ -164,13 +164,19 @@ GET  /debug/instance_resources
 GET  /debug/instance_loads
 ```
 
-Use `/debug/instance_loads` to inspect Proxy-maintained per-Instance `inflight` counters, `qps_1m` when present, queue-depth hints when the queue snapshot provider is available, and the queue-aware `least_load` score used for Instance selection.
+Use `/debug/instance_loads` to inspect Proxy-maintained per-Instance `inflight` counters, `qps_1m` when present, queue-pressure hints when the queue snapshot provider is available, and the queue-aware `least_load` score used for Instance selection.
 
 ```bash
 curl -sS http://127.0.0.1:8002/debug/instance_loads | python3 -m json.tool
 ```
 
-`least_load` uses a conservative deterministic score. `load.inflight` remains the primary signal with implicit weight `1.0`. When queue hints are available, the strategy also adds per-Instance active prepare work, active ready work, prepare queue depth, and ready queue depth. `qps_1m` from Instance heartbeats is supported but disabled by default with weight `0.0`. Missing metrics remain unavailable; they are not converted to measured zeroes, and the strategy falls back to round-robin if no load signal is known. Ties at the minimum score are also broken by round-robin.
+`least_load` uses a conservative deterministic score with three load-signal layers:
+
+1. Proxy lifecycle pressure: `inflight`.
+2. Instantaneous queue pressure: `active_prepare`, `active_ready`, `prepare_queue_depth`, and `ready_queue_depth`.
+3. Predicted reservation pressure from QueueManager timelines: `pending_prefill_count`, `active_decode_count`, and `predicted_total_backlog_ms`.
+
+`load.inflight` remains the primary signal with implicit weight `1.0`. Queue hints add per-Instance active prepare work, active ready work, prepare queue depth, ready queue depth, pending prefill reservations, active decode reservations, and predicted backlog time. `qps_1m` from Instance heartbeats is supported but disabled by default with weight `0.0`. Missing predicted fields remain unavailable and fall back to the instantaneous queue-aware behavior; missing queue hints fall back to inflight-only scoring. If no load signal is known for any candidate, selection falls back to round-robin. Ties at the minimum score are also broken by round-robin.
 
 Default `least_load` weights are:
 
@@ -181,9 +187,12 @@ Default `least_load` weights are:
 | `active_ready` | `1.0` | Proxy queue manager per-Instance snapshot |
 | `prepare_queue_depth` | `0.25` | Proxy queue manager per-Instance snapshot |
 | `ready_queue_depth` | `0.25` | Proxy queue manager per-Instance snapshot |
+| `pending_prefill_count` | `1.0` | Proxy QueueManager reservation snapshot |
+| `active_decode_count` | `0.5` | Proxy QueueManager reservation snapshot |
+| `predicted_total_backlog_ms` | `0.001` | Proxy QueueManager reservation snapshot |
 | `qps_1m` | `0.0` | Instance heartbeat |
 
-The debug response includes `least_load_score.total`, raw `least_load_score.components`, weighted components, metric sources, queue source availability, and the weights used to compute the score.
+The debug response includes the raw pressure fields (`pending_prefill_count`, `active_decode_count`, `next_slot_ready_in_ms`, `prefill_free_in_ms`, and `predicted_total_backlog_ms`) plus `least_load_score.total`, raw `least_load_score.components`, weighted components, metric sources, queue source availability, and the weights used to compute the score.
 
 The reporting path is:
 

proxy/proxy.py

@@ -96,7 +96,7 @@ async def _build_proxy_topology_meta() -> Dict[str, Any]:
 
 def _build_pool_resource_snapshot(app: FastAPI) -> Dict[str, Any]:
     pool: InstancePool = app.state.instance_pool  # type: ignore
-    queue_depths = queue_mgr.queue_depth_snapshot()
+    queue_depths = queue_mgr.queue_pressure_snapshot()
     return pool.build_pool_resource_snapshot(
         proxy_id=PROXY_ID,
         capacity=PROXY_MAX_CAPACITY,
@@ -134,7 +134,7 @@ async def lifespan(app: FastAPI):
     app.state.instance_pool = InstancePool(ttl_s=ttl_s)  # type: ignore
     p_control_plane.set_pool(app.state.instance_pool)  # type: ignore
     p_control_plane.set_pool_resource_context(PROXY_ID, PROXY_MAX_CAPACITY)
-    p_control_plane.set_queue_snapshot_provider(lambda: queue_mgr.queue_depth_snapshot())
+    p_control_plane.set_queue_snapshot_provider(lambda: queue_mgr.queue_pressure_snapshot())
 
     # --- Load the proxy scheduling strategy for the data plane ---
     strategy_name = os.environ.get("PROXY_INSTANCE_STRATEGY", "round_robin")
@@ -453,7 +453,7 @@ def select_instance(app: FastAPI, req_obj: SchedulerRequest):
 
     hint = {"request": req_obj}
     try:
-        hint["queue_depths"] = queue_mgr.queue_depth_snapshot()
+        hint["queue_depths"] = queue_mgr.queue_pressure_snapshot()
     except Exception:
         logger.warning(
             "[Proxy] queue snapshot failed during instance select; least_load will use inflight-only metrics",

proxy/queue/manager.py

@@ -99,12 +99,58 @@ def __init__(self) -> None:
 
 
     def queue_depth_snapshot(self) -> Dict[str, Any]:
-        """Return coarse queue depths without exposing task payloads.
+        """Return queue pressure without exposing task payloads.
 
-        Scheduler-facing pool_resource uses only the global totals. The per-instance
-        section stays Proxy-local for debugging.
+        Scheduler-facing pool_resource uses only the global instantaneous totals.
+        The per-instance section stays Proxy-local for debugging and Instance
+        selection. It combines instantaneous queue depths with predicted pressure
+        derived from reservation timelines.
+        """
+        return self.queue_pressure_snapshot()
+
+    def queue_pressure_snapshot(self) -> Dict[str, Any]:
+        """Return per-instance instantaneous and predicted queue pressure.
+
+        The snapshot intentionally exposes only aggregate numeric fields. It never
+        includes task payloads, request bodies, prompts, or knowledge content.
         """
         per_instance = self._qmap.snapshot_depths()
+        now_s = time.time()
+
+        instance_ids = set(per_instance.keys())
+        instance_ids.update(self._instance_pending_tasks.keys())
+        instance_ids.update(self._instance_decode_tasks.keys())
+        instance_ids.update(self._instance_slot_free_ts_s.keys())
+        instance_ids.update(self._instance_prefill_free_ts_s.keys())
+
+        for instance_id in instance_ids:
+            item = per_instance.setdefault(
+                instance_id,
+                {
+                    "prepare_queue_depth": 0,
+                    "ready_queue_depth": 0,
+                    "active_prepare": 0,
+                    "active_ready": 0,
+                },
+            )
+            slot_free_ts = self._instance_slot_free_ts_s.get(instance_id) or []
+            next_slot_ready_in_ms = 0.0
+            if slot_free_ts:
+                next_slot_ready_in_ms = max(0.0, min(slot_free_ts) - now_s) * 1000.0
+            prefill_free_in_ms = max(
+                0.0,
+                float(self._instance_prefill_free_ts_s.get(instance_id, now_s)) - now_s,
+            ) * 1000.0
+            item.update(
+                {
+                    "pending_prefill_count": int(len(self._instance_pending_tasks.get(instance_id, []))),
+                    "active_decode_count": int(len(self._instance_decode_tasks.get(instance_id, []))),
+                    "next_slot_ready_in_ms": float(next_slot_ready_in_ms),
+                    "prefill_free_in_ms": float(prefill_free_in_ms),
+                    "predicted_total_backlog_ms": float(max(next_slot_ready_in_ms, prefill_free_in_ms)),
+                }
+            )
+
         prepare_total = sum(item["prepare_queue_depth"] for item in per_instance.values())
         ready_total = sum(item["ready_queue_depth"] for item in per_instance.values())
         return {

proxy/resource/instance_pool.py

@@ -206,6 +206,11 @@ def snapshot_instance_loads(
                 "ready_queue_depth": queue_item.get("ready_queue_depth"),
                 "active_prepare": queue_item.get("active_prepare"),
                 "active_ready": queue_item.get("active_ready"),
+                "pending_prefill_count": queue_item.get("pending_prefill_count"),
+                "active_decode_count": queue_item.get("active_decode_count"),
+                "next_slot_ready_in_ms": queue_item.get("next_slot_ready_in_ms"),
+                "prefill_free_in_ms": queue_item.get("prefill_free_in_ms"),
+                "predicted_total_backlog_ms": queue_item.get("predicted_total_backlog_ms"),
                 "least_load_score": least_load_score,
             })
 

proxy/strategy/least_load.py

@@ -17,6 +17,9 @@ class LeastLoadWeights:
     active_ready: float = 1.0
     prepare_queue_depth: float = 0.25
     ready_queue_depth: float = 0.25
+    pending_prefill_count: float = 1.0
+    active_decode_count: float = 0.5
+    predicted_total_backlog_ms: float = 0.001
     qps_1m: float = 0.0
 
 
@@ -26,9 +29,10 @@ class LeastLoadStrategy(BaseInstanceStrategy):
 
     ``load.inflight`` remains the primary signal with implicit weight 1.0.
     Queue hints are added when the Proxy queue manager provides per-Instance
-    measurements. Missing metrics stay unknown rather than becoming measured
-    zeros. If no candidate has any measurable score component, selection falls
-    back to round-robin.
+    measurements, including instantaneous queue depth and predicted reservation
+    pressure. Missing metrics stay unknown rather than becoming measured zeros.
+    If no candidate has any measurable score component, selection falls back to
+    round-robin.
     """
 
     name = "least_load"
@@ -63,7 +67,15 @@ def compute_score(self, instance: InstanceLike, hint: Optional[Any] = None) -> D
         queue_item = self._queue_info(instance, hint)
         queue_source = "proxy_queue_manager" if queue_item is not None else "unavailable"
         if queue_item is not None:
-            for key in ("active_prepare", "active_ready", "prepare_queue_depth", "ready_queue_depth"):
+            for key in (
+                "active_prepare",
+                "active_ready",
+                "prepare_queue_depth",
+                "ready_queue_depth",
+                "pending_prefill_count",
+                "active_decode_count",
+                "predicted_total_backlog_ms",
+            ):
                 value = self._number_from_mapping(queue_item, key)
                 if value is not None:
                     components[key] = value
@@ -92,6 +104,9 @@ def weights_dict(self) -> Dict[str, float]:
             "active_ready": self.weights.active_ready,
             "prepare_queue_depth": self.weights.prepare_queue_depth,
             "ready_queue_depth": self.weights.ready_queue_depth,
+            "pending_prefill_count": self.weights.pending_prefill_count,
+            "active_decode_count": self.weights.active_decode_count,
+            "predicted_total_backlog_ms": self.weights.predicted_total_backlog_ms,
             "qps_1m": self.weights.qps_1m,
         }