epp: keep average aggregation, document credit assumptions

Revert the utilization detector's pool aggregation from max back to the
average it had before this PR. The scrape-lag credit corrects each
endpoint's score before aggregation, so the fix holds under average; max
changes Saturation() semantics for other consumers and warrants its own
design, tracked separately.

Document the credit's assumptions in the README and Saturation doc comment:
it requires runningRequestsSpec to be mapped, a single EPP replica
(InFlightRequests is per-replica while waiting+running is engine-global),
and aggregated (non-P/D) pools, since the default producer holds the
prefill request counter until end-of-stream.

Signed-off-by: Angelo Ruocco <ang@zurich.ibm.com>

Author: ruocco

pkg/epp/flowcontrol/controller/internal/processor.go

@@ -362,7 +362,6 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 
 	pool := p.endpointCandidates.Locate(ctx, nil)
 	p.poolEmpty = len(pool) == 0
-
 	saturation := p.saturationDetector.Saturation(ctx, pool)
 
 	// Record pool saturation metric

pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization/README.md

@@ -18,11 +18,10 @@ It relies on a "roofline model", evaluating both the queue depth and the KV cach
 
     EndpointScore = max(QueueDepth / QueueThreshold, KVCacheUsage / KVCacheThreshold)
 
-The global pool saturation is the roofline across all candidate endpoints:
+The global pool saturation is the average of the endpoint scores:
 
-    PoolSaturation = Max(EndpointScore)
+    PoolSaturation = Average(EndpointScore)
 
-The pool is saturated as soon as its hottest endpoint is, so a single overloaded endpoint is not diluted by idle ones.
 *Note: Endpoints with missing or stale metrics are aggressively scored as 100% saturated.*
 
 **Scrape-lag compensation:** `WaitingQueueSize` is scraped on a poller while the Flow Controller's dispatch loop runs far faster, so between two scrapes the queue term is stale-low and the gate would let the controller over-dispatch. When an `inflight-load-producer` is configured, the queue term is corrected by the in-flight requests the scrape does not yet reflect:
@@ -32,6 +31,12 @@ The pool is saturated as soon as its hottest endpoint is, so a single overloaded
 
 `InFlightRequests` is the producer's per-endpoint counter, incremented at dispatch and decremented on request completion, so the correction already accounts for requests that returned since the last scrape. Endpoints without the attribute contribute zero credit.
 
+The correction rests on assumptions that hold for a single-writer, aggregated deployment:
+
+- **`RunningRequestsSize` must be populated.** The subtraction of running requests is only sound when the metrics mapping defines `runningRequestsSpec` (defaults exist for vLLM and SGLang). A custom mapping without it leaves `RunningRequestsSize` at 0, and the credit over-counts by the running count.
+- **Single EPP replica.** `InFlightRequests` is tracked per EPP replica, while `WaitingQueueSize + RunningRequestsSize` is engine-global. With multiple EPP replicas the credit floors at 0 and the compensation silently disappears (the same single-writer assumption the concurrency detector makes).
+- **Aggregated (non-P/D) pools.** With the default producer config the prefill profile's request counter is held until end-of-stream, so under P/D disaggregation a prefill endpoint shows phantom credit for the full decode duration of every request whose prefill it served. Correcting this requires releasing the prefill request counter at start-of-stream in the producer.
+
 ### Role in Scheduling (The Traffic Shaper)
 The detector implements the `Filter` interface to protect individual endpoints. It removes endpoints from candidate lists if their telemetry is stale, or if they exceed specific safety limits:
 

pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization/detector.go

@@ -134,16 +134,12 @@ func (d *Detector) TypedName() fwkplugin.TypedName {
 //
 // It returns an aggregate saturation signal where:
 //
-//	Saturation = Max(PodSaturationScore)   // hottest endpoint drives the pool
+//	Saturation = Average(PodSaturationScore)
 //
 // For each pod, the score is determined by the most constrained resource (Compute or Memory):
 //
 //	PodScore = Max(WaitingQueue / QueueThreshold, KVCacheUsage / KVCacheThreshold)
 //
-// Aggregation is the MAX across endpoints (roofline at the pool level): the pool
-// is saturated as soon as its hottest endpoint is, so a single overloaded
-// endpoint is not diluted by idle ones.
-//
 // # Scrape-lag compensation
 //
 // WaitingQueueSize is scraped on a poller (default 50ms) while the flow
@@ -157,21 +153,27 @@ func (d *Detector) TypedName() fwkplugin.TypedName {
 //
 //	credit = max(0, InFlightRequests - (WaitingQueueSize + RunningRequestsSize))
 //
-// This credit is added only to the queue-depth term (the fast, controllable
-// signal); the KV-cache term is left as measured. Endpoints without the
-// attribute contribute zero credit and fall back to the scraped queue depth.
+// The credit corrects each endpoint's score before aggregation, so it holds
+// under either aggregation shape. It is added only to the queue-depth term (the
+// fast, controllable signal); the KV-cache term is left as measured. Endpoints
+// without the attribute contribute zero credit and fall back to the scraped
+// queue depth.
+//
+// The compensation assumes aggregated (non-P/D) pools. Under P/D disaggregation
+// with the default producer config, a prefill endpoint's in-flight request count
+// stays elevated for the whole decode duration of every request whose prefill it
+// served, so its credit is inflated; see the package README.
 func (d *Detector) Saturation(_ context.Context, candidates []datalayer.Endpoint) float64 {
 	if len(candidates) == 0 {
 		return 1.0
 	}
 
-	var maxScore float64
+	var totalScore float64
 	for _, e := range candidates {
 		metrics := e.GetMetrics()
 
 		if metrics == nil || time.Since(metrics.UpdateTime) > d.config.MetricsStalenessThreshold {
-			// Stale/missing metrics are treated as fully saturated (conservative).
-			maxScore = max(maxScore, 1.0)
+			totalScore += 1.0
 			continue
 		}
 
@@ -186,10 +188,10 @@ func (d *Detector) Saturation(_ context.Context, candidates []datalayer.Endpoint
 		kvRatio := metrics.KVCacheUsagePercent / d.config.KVCacheUtilThreshold
 
 		// Roofline Analysis: The pod is saturated if either resource is exhausted.
-		maxScore = max(maxScore, max(qRatio, kvRatio))
+		totalScore += max(qRatio, kvRatio)
 	}
 
-	return maxScore
+	return totalScore / float64(len(candidates))
 }
 
 // Filter blocks traffic to specific pods that are physically saturated or exceeding their safety limits.

pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization/detector_test.go

@@ -217,8 +217,8 @@ func TestDetector_Saturation(t *testing.T) {
 				// Pod2: Q=0/5(0.0), KV=0.2/0.9(0.22). Max=0.22...
 				makePodMetric("pod2", 0, 0.2, baseTime),
 			},
-			// Max(0.2, 0.222...) = 0.222...  (pool = hottest endpoint)
-			wantSaturation: 0.2 / 0.9,
+			// Avg(0.2, 0.222...) = 0.2111...
+			wantSaturation: (0.2 + (0.2 / 0.9)) / 2.0,
 		},
 		{
 			name: "Multiple pods, one good, one stale",
@@ -228,8 +228,8 @@ func TestDetector_Saturation(t *testing.T) {
 				// Pod2 (Stale): 1.0.
 				makePodMetric("pod2", 0, 0.2, baseTime.Add(-300*time.Millisecond)),
 			},
-			// Max(0.2, 1.0) = 1.0
-			wantSaturation: 1.0,
+			// Avg(0.2, 1.0) = 0.6
+			wantSaturation: 0.6,
 		},
 		{
 			name: "Multiple pods, one good, one bad (high queue)",
@@ -239,8 +239,8 @@ func TestDetector_Saturation(t *testing.T) {
 				// Pod2 (Bad): Q=15/5(3.0). Max=3.0.
 				makePodMetric("pod2", 15, 0.2, baseTime),
 			},
-			// Max(0.2, 3.0) = 3.0  (hottest endpoint drives the pool)
-			wantSaturation: 3.0,
+			// Avg(0.2, 3.0) = 1.6
+			wantSaturation: 1.6,
 		},
 		{
 			name: "Multiple pods, all bad capacity",
@@ -252,8 +252,8 @@ func TestDetector_Saturation(t *testing.T) {
 				// Pod3 (High KV): 0.99/0.90 = 1.1
 				makePodMetric("pod3", 1, 0.99, baseTime),
 			},
-			// Max(1.0, 4.0, 1.1) = 4.0
-			wantSaturation: 4.0,
+			// Avg(1.0, 4.0, 1.1) = 6.1 / 3 = 2.033...
+			wantSaturation: (1.0 + 4.0 + 1.1) / 3.0,
 		},
 		{
 			name: "Queue depth exactly at threshold",