TTFT aware Latency Scorer

cmd/runner/runner.go

@@ -55,6 +55,7 @@ import (
 	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/picker/weightedrandom"
 	inflightrequestsscorer "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/scorer/inflightrequests"
 	sessionaffinity "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/scorer/sessionaffinity"
+	ttftawarescorer "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/scorer/ttftaware"
 	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/requesthandling/basemodelextractor"
 	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/requesthandling/bodyfieldtoheader"
 	modelselectorplugin "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/requesthandling/modelselector"
@@ -301,6 +302,7 @@ func (r *Runner) registerInTreePlugins() {
 	plugin.Register(weightedrandom.WeightedRandomPickerType, weightedrandom.WeightedRandomPickerFactory)
 	plugin.Register(modelselectorplugin.ModelSelectorPluginType, modelselectorplugin.ModelSelectorPluginFactory)
 	plugin.Register(inflightrequestsscorer.PluginType, inflightrequestsscorer.ScorerFactory)
+	plugin.Register(ttftawarescorer.PluginType, ttftawarescorer.ScorerFactory)
 	plugin.Register(sessionaffinity.PluginType, sessionaffinity.ScorerFactory)
 	plugin.Register(modelnametoheader.PluginType, modelnametoheader.PluginFactory)
 }

pkg/framework/plugins/modelselector/scorer/ttftaware/README.md

@@ -0,0 +1,119 @@
+# TTFT-Aware Scorer
+
+Routes each request to the model with the lowest predicted TTFT under current load.
+
+It consumes the per-model snapshot published by the
+[TTFT percentile extractor](../../../datalayer/ttftpercentile/README.md) — the service floor
+`P10Low`, the operating points `LowTTFT` / `HighTTFT` (default P25 / P50, configured on the
+extractor) and their banded inflight averages `InflightAtLow` / `InflightAtHigh` — and turns them
+into a prediction and a score.
+
+## Prediction — `predictedTTFT`
+
+Every pool has a latency curve: how long a new request waits as a function of how many requests are
+already in flight. We measure three points on it and interpolate.
+
+| point | coordinates | meaning |
+|---|---|---|
+| **A** | `(0, P10Low)` | queue-free service time |
+| **B** | `(InflightAtLow, LowTTFT)` | low operating point (default P25) |
+| **C** | `(InflightAtHigh, HighTTFT)` | high operating point (default P50) |
+
+**A** and **C** always define the curve. **B** is inserted between them only when
+[admissible](#when-the-low-point-is-used), splitting it into two segments. Any single prediction
+reads one segment, so it uses two of the three points. Every point is something the extractor observed.
+
+```
+if B admissible:
+    if inflight < InflightAtLow:                 # segment A->B
+        predictedTTFT = P10Low + inflight * (LowTTFT - P10Low) / InflightAtLow
+    else:                                        # segment B->C, extended past C
+        predictedTTFT = LowTTFT + (inflight - InflightAtLow) *
+                        (HighTTFT - LowTTFT) / (InflightAtHigh - InflightAtLow)
+else:                                            # segment A->C, extended past C
+    predictedTTFT = P10Low + inflight * (HighTTFT - P10Low) / InflightAtHigh
+```
+
+The curve is continuous (both branches give `LowTTFT` at `InflightAtLow`), equals `P10Low` at zero
+load, and is monotone non-decreasing given `P10Low < LowTTFT < HighTTFT` — which the admissibility
+checks enforce. `predictedTTFT` is clamped to `>= P10Low` as a defensive guard.
+
+**Why three points and not two.** TTFT rises *faster than linearly* with inflight as a pool
+approaches saturation. A single chord from the floor to **C** cuts across that convex curve and
+under-predicts in between; **B** lets the curve bend so the loaded segment follows the local slope
+where the pool is actually operating.
+
+**Why A→B is a segment of its own.** The B→C slope is measured where queueing dominates, so it is
+steep. Running it backwards below **B** makes TTFT cross below the floor within a handful of
+requests — a draining-but-still-loaded pool would be predicted at its idle latency and win every
+decision it appeared in. Interpolating from `(0, floor)` matches how TTFT flattens as the queue
+drains.
+
+### When the low point is used
+
+**B** is admissible only when all of these hold. They are conditions on a measured point, not
+tuning knobs:
+
+| check | condition | why |
+|---|---|---|
+| separated in load | `InflightAtHigh - InflightAtLow >= minInflightGap` | the B→C slope is `ΔTTFT / Δinflight`; if both points sit at the same load that denominator is noise and the slope is meaningless |
+| ordered in latency | `HighTTFT > LowTTFT` | TTFT must rise with the percentile. If noise inverts them the slope goes negative and the *most* loaded pool scores best, feeding a saturated pool |
+| above the floor | `LowTTFT > P10Low` | `P10Low` is a long-window statistic and `LowTTFT` a recent one, so after a drain the recent P25 can fall below it — which would tilt the A→B segment downwards |
+| positive inflight | `InflightAtLow > 0` | keeps the A→B divisor safe; the gap check alone does not imply it |
+
+When **B** is dropped the curve is the single floor chord **A → C**, which is well defined at any
+load.
+
+
+### Model states
+
+- **cold** — `Floor() == 0` (never observed, or fewer than `minRequests` observations so the floor
+  is not yet trustworthy). Seeded optimistically at the best observed TTFT.
+- **seed** — has a floor but is not calibrated (`RecentN < minRequests`, or no inflight operating
+  point). Predicts at the floor.
+- **trusted** — `RecentN >= minRequests`, `InflightAtHigh > 0`, `HighTTFT > floor`. Uses the curve
+  above.
+
+## Score
+
+```
+score = (maxTTFT - predictedTTFT) / (maxTTFT - minTTFT)
+```
+
+Lowest predicted TTFT scores highest. Cold models seed at `minTTFT`; if every model is cold, all
+score 1.0.
+
+### Exploration
+
+An under-observed pool can be starved: competing against a calibrated pool it may never win the
+traffic it needs to calibrate. `explorationRate` breaks that loop — each under-observed pool is
+flipped independently per request, and with probability `explorationRate` its final score is forced
+to `1.0` so the picker sends it a probe; otherwise it is suppressed to `0`, but only when a
+calibrated pool exists to take the traffic. The override applies to the **final score only**, so a
+probe never distorts the trusted pools' normalisation.
+
+Together with the extractor's floor sample guard this reproduces what a manual warmup would do: a
+new pool reads as cold → receives probes → crosses `minRequests` → competes on its true latency.
+
+## Parameters
+
+| Parameter | Default | Description |
+|---|---|---|
+| `explorationRate` | 0.0 | Per-pool probability that an under-observed pool is probed on a given request. 0 = all traffic to the trusted winner. |
+| `minInflightGap` | 2.0 | Minimum inflight separation between the operating points for the low one to be used as an anchor. Must be > 0. |
+| `roundTTFTStep` | 0.0 | Quantize each prediction to a multiple of this many seconds before ranking (e.g. `0.01` = 10 ms). Pools landing in the same bucket tie and the picker splits them, instead of one winning on a difference too small to be meaningful. `0` = disabled. Must be >= 0. |
+
+`minRequests` is read from the extractor's published metrics and configured
+[there](../../../datalayer/ttftpercentile/README.md), not here.
+
+## Configuration
+
+```yaml
+- type: ttft-aware-scorer
+  parameters:
+    explorationRate: 0.1
+    minInflightGap: 2.0
+```
+
+The scorer requires `ttft-percentile-extractor` in `datalayer.extractors`, a model list (e.g. via
+`model-config-datasource`) so model selection has candidates, and a picker.

pkg/framework/plugins/modelselector/scorer/ttftaware/plugin.go

@@ -0,0 +1,247 @@
+/*
+Copyright 2026 The llm-d Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package ttftaware routes each request to the model with the lowest predicted TTFT under current
+// load. The prediction is a piecewise-linear curve through the model's measured points, evaluated
+// at its current inflight count. See README.md for the model, the equations and the rationale.
+package ttftaware
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math"
+	"math/rand"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	logutil "github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/logging"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/datalayer"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/modelselector"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/plugin"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/requesthandling"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/datalayer/ttftpercentile"
+)
+
+const (
+	PluginType = "ttft-aware-scorer"
+
+	defaultExplorationRate = 0.0 // disabled; 0.1 probes each under-observed model on ~10% of requests
+	defaultMinInflightGap  = 2.0 // inflight separation the operating points need to define a slope
+	defaultRoundTTFTStep   = 0.0 // disabled; e.g. 0.01 quantizes predictions to 10ms
+)
+
+var _ modelselector.Scorer = &TTFTAwareScorer{}
+
+// TTFTAwareScorerConfig holds optional parameters for the scorer plugin. See README.md.
+type TTFTAwareScorerConfig struct {
+	// ExplorationRate is the per-model probability that an under-observed model is forced to the
+	// top score, so it gets the traffic it needs to calibrate. Range [0, 1]. Default 0 (disabled).
+	ExplorationRate float64 `json:"explorationRate,omitempty"`
+	// MinInflightGap is the minimum inflight separation between the two operating points for the
+	// low one to be usable as a curve anchor. Below it their slope is noise. Must be > 0.
+	MinInflightGap float64 `json:"minInflightGap,omitempty"`
+	// RoundTTFTStep quantizes each prediction to a multiple of this many seconds before ranking
+	// (e.g. 0.01 = 10ms), so models closer together than the step tie and the picker splits them
+	// instead of one winning on a difference too small to be meaningful. Must be >= 0. Default 0.
+	RoundTTFTStep float64 `json:"roundTTFTStep,omitempty"`
+}
+
+type TTFTAwareScorer struct {
+	typedName       plugin.TypedName
+	explorationRate float64
+	minInflightGap  float64
+	roundTTFTStep   float64
+}
+
+func ScorerFactory(name string, parameters json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+	cfg := TTFTAwareScorerConfig{
+		ExplorationRate: defaultExplorationRate,
+		MinInflightGap:  defaultMinInflightGap,
+		RoundTTFTStep:   defaultRoundTTFTStep,
+	}
+	if len(parameters) > 0 {
+		if err := json.Unmarshal(parameters, &cfg); err != nil {
+			return nil, fmt.Errorf("failed to parse parameters for plugin %q: %w", name, err)
+		}
+	}
+	if cfg.ExplorationRate < 0 || cfg.ExplorationRate > 1 {
+		return nil, fmt.Errorf("explorationRate must be in [0, 1] for plugin %q", name)
+	}
+	if cfg.MinInflightGap <= 0 {
+		return nil, fmt.Errorf("minInflightGap must be > 0 for plugin %q", name)
+	}
+	if cfg.RoundTTFTStep < 0 {
+		return nil, fmt.Errorf("roundTTFTStep must be >= 0 for plugin %q", name)
+	}
+	return NewTTFTAwareScorer().
+		WithName(name).
+		WithExplorationRate(cfg.ExplorationRate).
+		WithMinInflightGap(cfg.MinInflightGap).
+		WithRoundTTFTStep(cfg.RoundTTFTStep), nil
+}
+
+func NewTTFTAwareScorer() *TTFTAwareScorer {
+	return &TTFTAwareScorer{
+		typedName:       plugin.TypedName{Type: PluginType, Name: PluginType},
+		explorationRate: defaultExplorationRate,
+		minInflightGap:  defaultMinInflightGap,
+		roundTTFTStep:   defaultRoundTTFTStep,
+	}
+}
+
+func (s *TTFTAwareScorer) TypedName() plugin.TypedName { return s.typedName }
+
+func (s *TTFTAwareScorer) WithName(name string) *TTFTAwareScorer {
+	s.typedName.Name = name
+	return s
+}
+
+func (s *TTFTAwareScorer) WithExplorationRate(r float64) *TTFTAwareScorer {
+	s.explorationRate = r
+	return s
+}
+
+func (s *TTFTAwareScorer) WithMinInflightGap(g float64) *TTFTAwareScorer {
+	s.minInflightGap = g
+	return s
+}
+
+func (s *TTFTAwareScorer) WithRoundTTFTStep(step float64) *TTFTAwareScorer {
+	s.roundTTFTStep = step
+	return s
+}
+
+// modelEval is the scorer's per-model working state for one Score call.
+type modelEval struct {
+	pred     float64
+	trusted  bool // calibrated operating point
+	observed bool // has a service floor (not truly cold)
+}
+
+// Score ranks models by predicted TTFT: score = (maxTTFT - predictedTTFT) / (maxTTFT - minTTFT).
+// Cold models are seeded at the best observed TTFT; if every model is cold, all score 1.0.
+func (s *TTFTAwareScorer) Score(ctx context.Context, _ *plugin.CycleState, _ *requesthandling.InferenceRequest, models []datalayer.Model) map[datalayer.Model]float64 {
+	evals := make([]modelEval, len(models))
+	minPred := math.MaxFloat64
+	maxPred := 0.0
+	anyObserved := false
+	anyTrusted := false
+
+	for i, model := range models {
+		pred, trusted, observed := s.predict(metricsFor(model))
+		if s.roundTTFTStep > 0 {
+			// Quantize before ranking so predictions closer together than the step tie.
+			pred = math.Round(pred/s.roundTTFTStep) * s.roundTTFTStep
+		}
+		evals[i] = modelEval{pred: pred, trusted: trusted, observed: observed}
+		if observed {
+			anyObserved = true
+			minPred = min(minPred, pred)
+			maxPred = max(maxPred, pred) // cold models seed to minPred, so they can never be the max
+		}
+		if trusted {
+			anyTrusted = true
+		}
+	}
+
+	scores := make(map[datalayer.Model]float64, len(models))
+
+	// No model has a floor yet -> nothing to rank; explore all equally.
+	if !anyObserved {
+		for _, model := range models {
+			scores[model] = 1.0
+		}
+		return scores
+	}
+
+	for i, model := range models {
+		e := &evals[i]
+		if !e.observed {
+			e.pred = minPred // seed cold models at the best observed TTFT (optimistic)
+		}
+		if maxPred == minPred {
+			scores[model] = 1.0
+		} else {
+			scores[model] = (maxPred - e.pred) / (maxPred - minPred)
+		}
+		// Independent coin per under-observed model: heads forces a calibration probe, tails
+		// suppresses it so calibrated models take the traffic. Final score only — never feeds the
+		// normalization above, so a probe cannot distort the trusted models' scores.
+		if s.explorationRate > 0 && !e.trusted {
+			if rand.Float64() < s.explorationRate {
+				scores[model] = 1.0
+			} else if anyTrusted {
+				scores[model] = 0
+			}
+		}
+	}
+
+	if dl := log.FromContext(ctx).V(logutil.DEBUG); dl.Enabled() {
+		for i, model := range models {
+			dl.Info("ttft-aware score", "model", model.GetName(),
+				"predictedTTFT", evals[i].pred, "score", scores[model], "trusted", evals[i].trusted)
+		}
+	}
+
+	return scores
+}
+
+// predict returns the model's predicted TTFT at its current inflight, and whether it is trusted
+// (calibrated) and observed (has a floor). Cold -> (0, false, false); observed but uncalibrated ->
+// the floor as an optimistic seed. See README.md for the curve and the admissibility checks.
+func (s *TTFTAwareScorer) predict(m ttftpercentile.TTFTPercentileMetrics) (pred float64, trusted, observed bool) {
+	floor := m.Floor()
+	if floor == 0 {
+		return 0, false, false
+	}
+	if !(m.RecentN >= m.MinRequests && m.InflightAtHigh > 0 && m.HighTTFT > floor) {
+		return floor, false, true
+	}
+
+	// The low point is admissible only if the two operating points are separated in load, ordered
+	// in latency, and above the floor — otherwise its segment slopes on noise or downwards.
+	useLow := m.InflightAtHigh-m.InflightAtLow >= s.minInflightGap &&
+		m.HighTTFT > m.LowTTFT &&
+		m.LowTTFT > floor &&
+		m.InflightAtLow > 0
+
+	cur := float64(m.Requests)
+	switch {
+	case useLow && cur < m.InflightAtLow:
+		// (0, floor) -> low point. Extending the steeper high-load slope backwards instead would
+		// predict a draining-but-loaded pool at its idle latency.
+		pred = floor + cur*(m.LowTTFT-floor)/m.InflightAtLow
+	case useLow:
+		// low -> high point, extended beyond it.
+		pred = m.LowTTFT + (cur-m.InflightAtLow)*(m.HighTTFT-m.LowTTFT)/(m.InflightAtHigh-m.InflightAtLow)
+	default:
+		// Low point inadmissible: floor chord (0, floor) -> high point, extended beyond it.
+		pred = floor + cur*(m.HighTTFT-floor)/m.InflightAtHigh
+	}
+	return max(pred, floor), true, true
+}
+
+// metricsFor reads the TTFT percentile metrics an extractor published for the model.
+// A missing or malformed attribute yields the zero value, which reads as truly cold.
+func metricsFor(model datalayer.Model) ttftpercentile.TTFTPercentileMetrics {
+	if val, err := datalayer.ReadAttributeKey[ttftpercentile.TTFTPercentileMetrics](
+		model.GetAttributes(), ttftpercentile.AttributeKey,
+	); err == nil {
+		return val
+	}
+	return ttftpercentile.TTFTPercentileMetrics{}
+}