Add native disaggregation controller

Introduces pkg/epp/disaggregation: a single Controller struct that owns
label-aware routing across the roles of a disaggregated inference
deployment (e.g. prefill / decode). Implements scheduling.Scorer and
requestcontrol.ResponseHeaderProcessor directly; Filter is done via two
thin wrappers installed by WireInto so strict-mode selectors run at the
head of the profile's filter chain and prefer-mode selectors at the
tail.

Config surface — one top-level disaggregation: YAML block:

```yaml
disaggregation:
  enabled: true
  scope:
    labelSelector: "disaggregatedset.x-k8s.io/name=my-set"
  selectors:
    - name: revision
      headerName: x-disagg-revision
      labelKey: disaggregatedset.x-k8s.io/revision
      mode: strict            # strict | prefer
  scoring:
    mode: sum                 # sum | disabled
    requireRoles:             # required when mode: sum
      labelKey: disaggregatedset.x-k8s.io/role
      values: [prefill, decode]
```

Nearly the same file works on every role's EPP; only scoring.mode
differs — sum on the originating side (prefill), disabled on the
terminal side (decode) where the filter has already pinned traffic to
a single revision and per-pod scoring adds nothing.

Why strict revision routing matters (not just a nice-to-have):
* disaggregated inference technically routes fine without any of this —
DisaggregatedSet's placement policy runs at a layer below routing, and
you could argue routing across revisions during a rolling update is
harmless. In production it isn't. When rolling from revision A to B,
you run into two failure classes:
  * Revision B not backward-compatible with A — either new-revision pods
    die outright when they receive a request meant for the old shape, or
    they fail gracefully and the retry also fails, dropping requests and
    tanking QoS.
  * Corrupted KVCache on revision A — vLLM produced a bad KV cache on
    some requests on a bad version that had a bug, and you do not want
    that cache carried into the newer revision's pods.

Both are cases where cross-revision routing during rollout is
actively harmful, which is why the revision selector runs in strict
mode. I have other examples, like when vLLM init past prefill and keep it
in its memory, but this is got solved.

At the NVL72 level (stride) IB connections still exist between
NVL72 racks, so prefer-mode is the right knob for that selector.

Signed-off-by: Mathis Felardos <mathis@mistral.ai>

Author: hasB4K

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -58,6 +58,13 @@ type EndpointPickerConfig struct {
 	// This configuration is only respected if the "flowControl" FeatureGate is enabled.
 	FlowControl *FlowControlConfig `json:"flowControl,omitempty"`
 
+	// +optional
+	// Disaggregation configures native, label-aware routing across roles of a
+	// disaggregated inference deployment (e.g. prefill / decode). Parsed by
+	// pkg/epp/disaggregation.Register at boot; opaque here to avoid coupling
+	// the schema to the implementation.
+	Disaggregation *json.RawMessage `json:"disaggregation,omitempty"`
+
 	// +optional
 	// RequestHandler specifies the handling logic used by the EPP to process incoming requests.
 	RequestHandler *RequestHandlerConfig `json:"requestHandler,omitempty"`

apix/config/v1alpha1/zz_generated.deepcopy.go

@@ -0,0 +1,52 @@
+// Copyright 2025 The Kubernetes 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
+
+package runner
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+
+	"github.com/llm-d/llm-d-router/pkg/epp/disaggregation"
+)
+
+// registerDisaggregation instantiates the disaggregation controller if the
+// rawConfig contains a disaggregation block and hands it to WireInto, which
+// does the correct-position registration in the scheduler and request-control
+// pipelines. No-op when the block is absent.
+func (r *Runner) registerDisaggregation(ctx context.Context, restCfg *rest.Config, namespace string) error {
+	if r.rawConfig == nil || r.rawConfig.Disaggregation == nil {
+		return nil
+	}
+
+	var config disaggregation.Config
+	if err := json.Unmarshal(*r.rawConfig.Disaggregation, &config); err != nil {
+		return fmt.Errorf("parse disaggregation config: %w", err)
+	}
+	if !config.Enabled {
+		return nil
+	}
+
+	client, err := kubernetes.NewForConfig(restCfg)
+	if err != nil {
+		return fmt.Errorf("build kubernetes client for disaggregation: %w", err)
+	}
+
+	controller, err := disaggregation.Register(ctx, client, namespace, config)
+	if err != nil {
+		return fmt.Errorf("register disaggregation controller: %w", err)
+	}
+	if controller == nil {
+		return nil
+	}
+	return disaggregation.WireInto(r.schedulerConfig, r.requestControlConfig, controller)
+}

cmd/epp/runner/disaggregation.go

@@ -439,6 +439,11 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 
 	setupLog.Info("parsed config", "scheduler-config", r.schedulerConfig)
 
+	if err := r.registerDisaggregation(ctx, cfg, gknn.Namespace); err != nil {
+		setupLog.Error(err, "failed to register disaggregation controller")
+		return nil, nil, err
+	}
+
 	scheduler := scheduling.NewSchedulerWithConfig(r.schedulerConfig)
 
 	if err := r.configureAndStartDatalayer(ctx, eppConfig.DataConfig, mgr); err != nil {

cmd/epp/runner/runner.go

@@ -0,0 +1,180 @@
+// Package disaggregation implements native, label-aware routing across roles
+// of a disaggregated inference deployment (e.g. prefill/decode).
+//
+// A single Controller implements scheduling.Scorer and
+// requestcontrol.ResponseHeaderProcessor. Filter is handled by two thin
+// wrappers built by WireInto: strict-mode selectors run at the head of the
+// scheduler profile's filter chain, prefer-mode selectors at the tail.
+//
+// Boot flow: Register (validate config, start pod cache, verify config
+// against observed cluster state) then WireInto (attach controller to
+// scheduler profiles and request-control pipeline). Both are called from
+// cmd/epp/runner/disaggregation.go.
+package disaggregation
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+// Config is the top-level disaggregation block loaded from the EPP YAML.
+type Config struct {
+	Enabled   bool       `json:"enabled"`
+	Scope     Scope      `json:"scope"`
+	Selectors []Selector `json:"selectors"`
+	Scoring   Scoring    `json:"scoring"`
+}
+
+// Scope constrains which pods the informer observes.
+type Scope struct {
+	LabelSelector string `json:"labelSelector"`
+	// Namespace defaults to the InferencePool's namespace when empty.
+	Namespace string `json:"namespace,omitempty"`
+}
+
+// Selector defines one header/label pair to filter and tag on.
+type Selector struct {
+	Name       string       `json:"name"`
+	HeaderName string       `json:"headerName"`
+	LabelKey   string       `json:"labelKey"`
+	Mode       SelectorMode `json:"mode"`
+}
+
+// UnmarshalJSON normalises HeaderName to lowercase at parse time. Downstream
+// code compares against `request.Headers[headerName]` which are lowercased
+// by the framework; keeping normalisation at the boundary means callers of
+// Validate see a pure check with no side effects.
+func (s *Selector) UnmarshalJSON(data []byte) error {
+	// Alias avoids infinite recursion into this method.
+	type raw Selector
+	var parsed raw
+	if err := json.Unmarshal(data, &parsed); err != nil {
+		return err
+	}
+	parsed.HeaderName = strings.ToLower(parsed.HeaderName)
+	*s = Selector(parsed)
+	return nil
+}
+
+// SelectorMode governs filter behaviour when the header carries a value with no
+// matching candidates.
+type SelectorMode string
+
+const (
+	// ModeStrict: no match → empty candidate set → framework returns 503.
+	ModeStrict SelectorMode = "strict"
+	// ModePrefer: no match → fall back to the unfiltered candidate set.
+	// (Escape-hatch semantics, not a spectrum; when at least one candidate
+	// matches, ModePrefer behaves identically to ModeStrict.)
+	ModePrefer SelectorMode = "prefer"
+)
+
+// Scoring picks the weighting strategy applied to candidates that survived filtering.
+type Scoring struct {
+	Mode         ScoringMode   `json:"mode"`
+	RequireRoles *RequireRoles `json:"requireRoles,omitempty"`
+}
+
+// ScoringMode selects the weighting strategy.
+type ScoringMode string
+
+const (
+	// ScoringModeSum weights each candidate uniformly (1.0 per pod), with the
+	// requireRoles gate excluding revisions where any listed role has zero
+	// Ready pods across scope. Paired with weighted-random-picker this gives
+	// pod-count-proportional traffic over the surviving revisions.
+	ScoringModeSum ScoringMode = "sum"
+	// ScoringModeDisabled skips the scorer entirely — the controller does not
+	// register as a Scorer at boot.
+	ScoringModeDisabled ScoringMode = "disabled"
+)
+
+// RequireRoles gates revisions on cross-role liveness: a revision is eligible for
+// scoring only when every listed role has at least one Ready pod for that revision.
+type RequireRoles struct {
+	LabelKey string   `json:"labelKey"`
+	Values   []string `json:"values"`
+}
+
+// Validate performs static config checks. A disabled config validates
+// trivially. All returned errors carry the offending field path so operators
+// can locate the problem quickly. Pure — no side effects; header
+// normalisation happens at JSON unmarshal time via Selector.UnmarshalJSON.
+func (c *Config) Validate() error {
+	if !c.Enabled {
+		return nil
+	}
+
+	if c.Scope.LabelSelector == "" {
+		return fmt.Errorf("disaggregation.scope.labelSelector is required when enabled")
+	}
+	if _, err := labels.Parse(c.Scope.LabelSelector); err != nil {
+		return fmt.Errorf("disaggregation.scope.labelSelector is not a valid label selector: %w", err)
+	}
+
+	if len(c.Selectors) == 0 {
+		return fmt.Errorf("disaggregation.selectors must contain at least one entry")
+	}
+	seenNames := make(map[string]struct{}, len(c.Selectors))
+	seenHeaders := make(map[string]struct{}, len(c.Selectors))
+	for index := range c.Selectors {
+		selector := &c.Selectors[index]
+		if selector.Name == "" {
+			return fmt.Errorf("disaggregation.selectors[%d].name is required", index)
+		}
+		if selector.HeaderName == "" {
+			return fmt.Errorf("disaggregation.selectors[%d].headerName is required", index)
+		}
+		if selector.LabelKey == "" {
+			return fmt.Errorf("disaggregation.selectors[%d].labelKey is required", index)
+		}
+		switch selector.Mode {
+		case ModeStrict, ModePrefer:
+		default:
+			return fmt.Errorf("disaggregation.selectors[%d].mode %q must be one of strict|prefer", index, selector.Mode)
+		}
+
+		if _, duplicate := seenNames[selector.Name]; duplicate {
+			return fmt.Errorf("disaggregation.selectors: duplicate selector name %q", selector.Name)
+		}
+		seenNames[selector.Name] = struct{}{}
+
+		if _, duplicate := seenHeaders[selector.HeaderName]; duplicate {
+			return fmt.Errorf("disaggregation.selectors: duplicate header name %q", selector.HeaderName)
+		}
+		seenHeaders[selector.HeaderName] = struct{}{}
+	}
+
+	switch c.Scoring.Mode {
+	case ScoringModeSum:
+		if c.Scoring.RequireRoles == nil {
+			return fmt.Errorf("disaggregation.scoring.requireRoles is required when scoring.mode=sum")
+		}
+		if c.Scoring.RequireRoles.LabelKey == "" {
+			return fmt.Errorf("disaggregation.scoring.requireRoles.labelKey is required")
+		}
+		if len(c.Scoring.RequireRoles.Values) == 0 {
+			return fmt.Errorf("disaggregation.scoring.requireRoles.values must contain at least one role")
+		}
+	case ScoringModeDisabled:
+	default:
+		return fmt.Errorf("disaggregation.scoring.mode %q must be one of sum|disabled", c.Scoring.Mode)
+	}
+
+	return nil
+}
+
+// HasSelectorsInMode reports whether any selector in this config runs in the
+// given mode. Used by wiring to skip registration of a strict/prefer wrapper
+// that would have nothing to do.
+func (c *Config) HasSelectorsInMode(mode SelectorMode) bool {
+	for _, selector := range c.Selectors {
+		if selector.Mode == mode {
+			return true
+		}
+	}
+	return false
+}