feat: Introduce pluggable inter-flow framework (kubernetes-sigs#1167)

This commit introduces the `InterFlowDispatchPolicy` framework, the third and final major component of the new pluggable flow control system. This framework decouples the logic for selecting which flow's queue to service next (fairness between flows) from the core controller.

This completes the two-tier policy model, where the `InterFlowDispatchPolicy` makes the strategic decision about which flow to service, and the `IntraFlowDispatchPolicy` makes the tactical decision of which request to select from within that flow's queue.

Key components include:
- `framework.InterFlowDispatchPolicy`: The core interface that defines the contract for selecting a queue from a priority band.
- `framework.PriorityBandAccessor`: A read-only interface that provides policies with safe access to the state of all queues within a priority level.
- A factory and registration system for discovering and instantiating policy plugins by name.
- A comprehensive functional conformance test suite to validate the contract for all policy plugins.
- A `roundrobin` policy for fair, sequential queue selection.
- A `besthead` policy for greedy, utilization-focused queue selection.

Author: LukeAVanDrie

pkg/epp/flowcontrol/framework/errors.go

@@ -42,3 +42,11 @@ var (
 	// provided, valid `types.QueueItemHandle`. This can occur if the item was removed by a concurrent operation.
 	ErrQueueItemNotFound = errors.New("queue item not found for the given handle")
 )
+
+// Policy Errors
+var (
+	// ErrIncompatiblePriorityType indicates that an `InterFlowDispatchPolicy` (like "BestHead") attempted to compare
+	// items from two different flow queues whose `ItemComparator`s have different `ScoreType` values, making a
+	// meaningful comparison impossible.
+	ErrIncompatiblePriorityType = errors.New("incompatible priority score type for comparison")
+)

pkg/epp/flowcontrol/framework/mocks/mocks.go

@@ -63,3 +63,42 @@ func (m *MockFlowQueueAccessor) Comparator() framework.ItemComparator { return m
 func (m *MockFlowQueueAccessor) FlowSpec() types.FlowSpecification    { return m.FlowSpecV }
 
 var _ framework.FlowQueueAccessor = &MockFlowQueueAccessor{}
+
+// MockPriorityBandAccessor is a mock implementation of the `framework.PriorityBandAccessor` interface.
+type MockPriorityBandAccessor struct {
+	PriorityV      uint
+	PriorityNameV  string
+	FlowIDsV       []string
+	QueueV         framework.FlowQueueAccessor // Value to return for any Queue(flowID) call
+	QueueFuncV     func(flowID string) framework.FlowQueueAccessor
+	IterateQueuesV func(callback func(queue framework.FlowQueueAccessor) bool)
+}
+
+func (m *MockPriorityBandAccessor) Priority() uint       { return m.PriorityV }
+func (m *MockPriorityBandAccessor) PriorityName() string { return m.PriorityNameV }
+func (m *MockPriorityBandAccessor) FlowIDs() []string    { return m.FlowIDsV }
+
+func (m *MockPriorityBandAccessor) Queue(flowID string) framework.FlowQueueAccessor {
+	if m.QueueFuncV != nil {
+		return m.QueueFuncV(flowID)
+	}
+	return m.QueueV
+}
+
+func (m *MockPriorityBandAccessor) IterateQueues(callback func(queue framework.FlowQueueAccessor) bool) {
+	if m.IterateQueuesV != nil {
+		m.IterateQueuesV(callback)
+	} else {
+		// Default behavior: iterate based on FlowIDsV and QueueV/QueueFuncV
+		for _, id := range m.FlowIDsV {
+			q := m.Queue(id)
+			if q != nil { // Only call callback if queue exists
+				if !callback(q) {
+					return
+				}
+			}
+		}
+	}
+}
+
+var _ framework.PriorityBandAccessor = &MockPriorityBandAccessor{}

pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/README.md

@@ -0,0 +1,57 @@
+# Flow Controller Inter-Flow Dispatch Policy Plugins
+
+This directory contains concrete implementations of the [`framework.InterFlowDispatchPolicy`](../../../policies.go)
+interface. These policies are responsible for determining *which flow's queue* gets the next opportunity to dispatch a
+request to the scheduler.
+
+## Overview
+
+The `controller.FlowController` uses a two-tier policy system to manage requests. `framework.InterFlowDispatchPolicy`
+plugins represent the second tier, making strategic decisions about fairness *between* different logical flows (e.g.,
+between different tenants or models).
+
+This contrasts with the `framework.IntraFlowDispatchPolicy`, which is responsible for **temporal scheduling**: deciding
+the order of requests *within* a single flow's queue after it has been selected by the inter-flow policy.
+
+Key responsibilities and characteristics of a `framework.InterFlowDispatchPolicy`:
+
+1.  **Queue Selection (`SelectQueue`)**: The primary method, `SelectQueue(band framework.PriorityBandAccessor)`,
+    inspects a set of queues within a single priority level (a "band") and decides which queue, if any, should be
+    selected to dispatch a request from next.
+
+2.  **Fairness Across Flows**: The core purpose of this policy is to enforce a fairness doctrine across multiple
+    competing flows. This could be simple round-robin, or more complex weighted fairness schemes.
+
+3.  **Stateless vs. Stateful**: Policies can be stateless (like `besthead`, which makes a decision based only on the
+    current state of the queues) or stateful (like `roundrobin`, which needs to remember which queue it selected last).
+    Any state must be managed in a goroutine-safe manner.
+
+The `framework.InterFlowDispatchPolicy` is critical for multi-tenancy and preventing any single high-traffic flow from
+starving all others.
+
+## Contributing a New `framework.InterFlowDispatchPolicy` Implementation
+
+To contribute a new dispatch policy implementation, follow these steps:
+
+1.  **Define Your Implementation**
+    - Create a new Go package in a subdirectory (e.g., `mycustompolicy/`).
+    - Implement the `framework.InterFlowDispatchPolicy` interface.
+    - Ensure all methods are goroutine-safe if your policy maintains any internal state.
+
+2.  **Register Your Policy**
+    - In an `init()` function within your policy's Go file, call [`MustRegisterPolicy()`](./factory.go) with a unique
+      name and a constructor function that matches the `PolicyConstructor` signature.
+
+3.  **Add to the Functional Test**
+    - Add a blank import for your new package to [`functional_test.go`](./functional_test.go). Your policy will then
+      be automatically included in the functional test suite, which validates the basic
+      `framework.InterFlowDispatchPolicy` contract (e.g., correct initialization, handling of nil/empty bands).
+
+4.  **Add Policy-Specific Tests**
+    - The functional test suite only validates the universal contract. You MUST add a separate `_test.go` file within
+      your package to test the specific logic of your policy.
+    - For example, your tests should validate that `SelectQueue()` correctly implements your desired selection logic
+      (e.g., that round-robin correctly cycles through queues).
+
+5.  **Documentation**
+    - Add a package-level GoDoc comment to your new policy's Go file, explaining its behavior and any trade-offs.

pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead.go

@@ -0,0 +1,99 @@
+/*
+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
+
+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 besthead provides a `framework.InterFlowDispatchPolicy` that selects the queue containing the single "best"
+// item from across all queues in a priority band.
+package besthead
+
+import (
+	"fmt"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
+)
+
+// BestHeadPolicyName is the name of the Best Head policy implementation.
+const BestHeadPolicyName = "BestHead"
+
+func init() {
+	dispatch.MustRegisterPolicy(dispatch.RegisteredPolicyName(BestHeadPolicyName),
+		func() (framework.InterFlowDispatchPolicy, error) {
+			return newBestHead(), nil
+		})
+}
+
+type bestHead struct{}
+
+func newBestHead() *bestHead {
+	return &bestHead{}
+}
+
+// Name returns the name of the policy.
+func (p *bestHead) Name() string {
+	return BestHeadPolicyName
+}
+
+// SelectQueue implements a greedy strategy that bypasses fairness concerns to select the queue containing the single
+// "best" item from across all queues in the priority band. It iterates through all non-empty queues, peeks at their
+// head items, and uses the `framework.ItemComparator` from each queue to find the highest-priority item overall.
+//
+// This policy is useful for maximizing utilization when fairness between flows is not a concern. It requires that all
+// queues being compared have a compatible `framework.ScoreType` to ensure the comparison is meaningful. If an
+// incompatible comparator is found, the selection fails with an error.
+func (p *bestHead) SelectQueue(band framework.PriorityBandAccessor) (framework.FlowQueueAccessor, error) {
+	if band == nil {
+		return nil, nil
+	}
+
+	var bestQueue framework.FlowQueueAccessor
+	var bestItem types.QueueItemAccessor
+
+	var iterationErr error
+	band.IterateQueues(func(queue framework.FlowQueueAccessor) (keepIterating bool) {
+		if queue == nil || queue.Len() == 0 {
+			return true
+		}
+
+		item, err := queue.PeekHead()
+		if err != nil || item == nil {
+			return true
+		}
+
+		if bestQueue == nil {
+			bestQueue = queue
+			bestItem = item
+			return true
+		}
+
+		if queue.Comparator().ScoreType() != bestQueue.Comparator().ScoreType() {
+			iterationErr = fmt.Errorf("%w: expected %q, got %q", framework.ErrIncompatiblePriorityType,
+				bestQueue.Comparator().ScoreType(), queue.Comparator().ScoreType())
+			return false
+		}
+
+		if bestQueue.Comparator().Func()(item, bestItem) {
+			bestQueue = queue
+			bestItem = item
+		}
+		return true
+	})
+
+	if iterationErr != nil {
+		return nil, iterationErr
+	}
+	return bestQueue, nil
+}