llm-d · LukeAVanDrie · Jul 23, 2026 · Jul 13, 2026 · Jul 16, 2026 · Jul 17, 2026
@@ -34,7 +34,7 @@ The plugin internally tracks active concurrency by hooking into the request life
 
 The plugin accepts JSON parameters decoding to the following fields:
 
-- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"` or `"tokens"`. (Default: `"requests"`)
+- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"`, `"tokens"`, or `"hybrid"`. In `"hybrid"` mode both request and token accounting are evaluated and the more constraining dimension wins: pool saturation is the larger of the two ratios, and an endpoint is filtered out when either its request load or its token load reaches the limit. (Default: `"requests"`)
 - `maxConcurrency` (`int64`): Maximum requests in flight. Serves as the "ideal" request capacity for a single endpoint. Must be > 0. (Default: `100`)
 - `maxTokenConcurrency` (`int64`): Maximum tokens in flight. The "tokens" mode equivalent of `maxConcurrency`. Must be > 0. (Default: `1000000`)
 - `headroom` (`float64`): Allowed burst capacity above the ideal threshold, expressed as a fraction (e.g., `0.2` for 20%). Must be >= 0.0. (Default: `0.0`)

@@ -62,6 +62,8 @@ type apiConfig struct {
 	// Valid values are:
 	// - "requests": use discrete request counts for capacity accounting.
 	// - "tokens": use estimated token counts for capacity accounting.
+	// - "hybrid": evaluate both request and token accounting and report whichever
+	//   reaches saturation first (the more constraining of the two).
 	//
 	// Defaults to "requests" if unset.
 	ConcurrencyMode *concurrencyMode `json:"concurrencyMode,omitempty"`
@@ -85,6 +87,8 @@ const (
 	modeRequests concurrencyMode = "requests"
 	// modeTokens uses token count for concurrency detection.
 	modeTokens concurrencyMode = "tokens"
+	// modeHybrid uses both request and token counts, reporting whichever saturates first.
+	modeHybrid concurrencyMode = "hybrid"
 )
 
 const (
@@ -164,7 +168,7 @@ func validateConfig(cfg *apiConfig) error {
 
 	if cfg.ConcurrencyMode != nil {
 		switch *cfg.ConcurrencyMode {
-		case modeRequests, modeTokens:
+		case modeRequests, modeTokens, modeHybrid:
 			// Valid
 		default:
 			errs = append(errs, fmt.Errorf("unsupported concurrencyMode: %q", *cfg.ConcurrencyMode))

@@ -121,49 +121,58 @@ func (d *detector) getLoad(m datalayer.AttributeMap) *attrconcurrency.InFlightLo
 
 // Saturation calculates the saturation level of the pool.
 //
-// It returns an aggregate saturation signal where:
+// It returns an aggregate saturation signal where each dimension is evaluated as:
 //
-//	Saturation = Total Inflight Requests / Total MaxConcurrency Capacity.
+//	Saturation = Total Inflight / Total Capacity.
+//
+// In "hybrid" mode both the request and token dimensions are computed and the larger
+// (more saturated) of the two is returned, so the pool reports saturated as soon as
+// either dimension is exhausted.
 func (d *detector) Saturation(_ context.Context, endpoints []datalayer.Endpoint) float64 {
 	if len(endpoints) == 0 {
 		return 1.0
 	}
 
-	var totalInflight, totalCapacity int64
+	var reqInflight, reqCapacity, tokInflight, tokCapacity int64
 	for _, e := range endpoints {
 		if e == nil {
 			continue
 		}
 
-		if d.config.mode == modeTokens {
-			totalCapacity += d.config.maxTokenConcurrency
-		} else {
-			totalCapacity += d.config.maxConcurrency
-		}
+		reqCapacity += d.config.maxConcurrency
+		tokCapacity += d.config.maxTokenConcurrency
 
 		if e.GetMetadata() == nil {
 			continue
 		}
 
 		load := d.getLoad(e.GetAttributes())
+		reqInflight += load.Requests
+		tokInflight += load.Tokens
+	}
 
-		if d.config.mode == modeTokens {
-			totalInflight += load.Tokens
-		} else {
-			totalInflight += load.Requests
-		}
+	switch d.config.mode {
+	case modeTokens:
+		return ratio(tokInflight, tokCapacity)
+	case modeHybrid:
+		return max(ratio(reqInflight, reqCapacity), ratio(tokInflight, tokCapacity))
+	default:
+		return ratio(reqInflight, reqCapacity)
 	}
+}
 
-	if totalCapacity == 0 {
+// ratio computes inflight/capacity, failing closed (1.0) when capacity is zero.
+func ratio(inflight, capacity int64) float64 {
+	if capacity == 0 {
 		return 1.0
 	}
-
-	return float64(totalInflight) / float64(totalCapacity)
+	return float64(inflight) / float64(capacity)
 }
 
 // Filter blocks traffic to specific endpoints that are physically saturated or exceeding their safety limits.
 //
-// It applies a relaxed limit (MaxConcurrency * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance.
+// It applies a relaxed limit (Capacity * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance.
+// In "hybrid" mode an endpoint is dropped when either its request load or its token load reaches the limit.
 func (d *detector) Filter(
 	_ context.Context,
 	_ *fwksched.InferenceRequest,
@@ -172,28 +181,30 @@ func (d *detector) Filter(
 	// Pre-allocate assuming most endpoints will pass the filter to minimize allocations.
 	filtered := make([]fwksched.Endpoint, 0, len(endpoints))
 
-	var limit int64
-	if d.config.mode == modeTokens {
-		limit = int64(float64(d.config.maxTokenConcurrency) * (1.0 + d.config.headroom))
-	} else {
-		limit = int64(float64(d.config.maxConcurrency) * (1.0 + d.config.headroom))
-	}
+	reqLimit := int64(float64(d.config.maxConcurrency) * (1.0 + d.config.headroom))
+	tokLimit := int64(float64(d.config.maxTokenConcurrency) * (1.0 + d.config.headroom))
 
 	for _, e := range endpoints {
 		if e == nil {
 			continue
 		}
 		load := d.getLoad(e)
 
-		if d.config.mode == modeTokens {
-			if load.Tokens < limit {
-				filtered = append(filtered, e)
-			}
-		} else {
-			if load.Requests < limit {
-				filtered = append(filtered, e)
-			}
+		if d.admits(load, reqLimit, tokLimit) {
+			filtered = append(filtered, e)
 		}
 	}
 	return filtered
 }
+
+// admits reports whether an endpoint is below its safety limit for the active mode.
+func (d *detector) admits(load *attrconcurrency.InFlightLoad, reqLimit, tokLimit int64) bool {
+	switch d.config.mode {
+	case modeTokens:
+		return load.Tokens < tokLimit
+	case modeHybrid:
+		return load.Requests < reqLimit && load.Tokens < tokLimit
+	default:
+		return load.Requests < reqLimit
+	}
+}
@@ -110,6 +110,11 @@
 			configJSON: []byte(`{"headroom": -0.5}`),
 			wantError:  true,
 		},
+		{
+			name:       "valid hybrid mode",
+			configJSON: []byte(`{"concurrencyMode": "hybrid", "maxConcurrency": 10, "maxTokenConcurrency": 100}`),
+			wantError:  false,
+		},
 		{
 			name:       "invalid mode",
 			configJSON: []byte(`{"concurrencyMode": "magic"}`),
@@ -490,6 +495,88 @@
 	require.InDelta(t, 0.0, detector.Saturation(ctx, candidates), 1e-6, "expected clean state after DeleteEndpoint")
 }
 
+// TestDetector_HybridSaturation verifies hybrid mode reports the more saturated dimension.
+func TestDetector_HybridSaturation(t *testing.T) {
+	t.Parallel()
+
+	config := config{
+		mode:                modeHybrid,
+		maxConcurrency:      10,
+		maxTokenConcurrency: 100,
+	}
+
+	tests := []struct {
+		name           string
+		requests       int64
+		tokens         int64
+		wantSaturation float64
+	}{
+		{name: "empty", requests: 0, tokens: 0, wantSaturation: 0.0},
+		{name: "tokens_dominate", requests: 2, tokens: 80, wantSaturation: 0.8},  // max(0.2, 0.8)
+		{name: "requests_dominate", requests: 9, tokens: 10, wantSaturation: 0.9}, // max(0.9, 0.1)
+		{name: "request_dimension_saturates_first", requests: 10, tokens: 5, wantSaturation: 1.0},
+		{name: "token_dimension_saturates_first", requests: 1, tokens: 100, wantSaturation: 1.0},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			reg := newLocalRegistry()
+			ctx := context.Background()
+			detector := newDetector("test-detector", config, logr.Discard())
+			endpointName := "endpoint-a"
+			reg.update(fullEndpointName(endpointName), func(load *attrconcurrency.InFlightLoad) {
+				load.Requests = tc.requests
+				load.Tokens = tc.tokens
+			})
+
+			got := detector.Saturation(ctx, []datalayer.Endpoint{newFakeEndpoint(reg, endpointName)})
+			require.InDelta(t, tc.wantSaturation, got, 1e-6, "hybrid saturation mismatch")
+		})
+	}
+}
+
+// TestDetector_HybridFilter verifies hybrid mode drops an endpoint when either dimension hits its limit.
+func TestDetector_HybridFilter(t *testing.T) {
+	t.Parallel()
+
+	// headroom 0.0 -> limits are exactly maxConcurrency (10) and maxTokenConcurrency (100).
+	config := config{
+		mode:                modeHybrid,
+		maxConcurrency:      10,
+		maxTokenConcurrency: 100,
+	}
+	ctx := context.Background()
+
+	tests := []struct {
+		name     string
+		requests int64
+		tokens   int64
+		wantKept int
+	}{
+		{name: "below_both_limits", requests: 5, tokens: 50, wantKept: 1},
+		{name: "request_limit_reached", requests: 10, tokens: 50, wantKept: 0},
+		{name: "token_limit_reached", requests: 5, tokens: 100, wantKept: 0},
+		{name: "both_over", requests: 20, tokens: 200, wantKept: 0},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			reg := newLocalRegistry()
+			detector := newDetector("test-detector", config, logr.Discard())
+			endpointName := "endpoint-a"
+			reg.update(fullEndpointName(endpointName), func(load *attrconcurrency.InFlightLoad) {
+				load.Requests = tc.requests
+				load.Tokens = tc.tokens
+			})
+
+			kept := detector.Filter(ctx, nil, []fwksched.Endpoint{newStubSchedulingEndpoint(reg, endpointName)})
+			require.Len(t, kept, tc.wantKept, "hybrid filter mismatch")
+		})
+	}
+}
+
 // TestDetector_ConcurrencyStress performs race condition check.
 func TestDetector_ConcurrencyStress(t *testing.T) {
 	t.Parallel()