Introduce pluggable intra-flow dispatch framework

This commit introduces the `IntraFlowDispatchPolicy` framework, the second major component of the new pluggable flow control system. This framework decouples the logic for selecting a request from within a single flow's queue (temporal scheduling) from the underlying queue data structure.

Key components include:
- `framework.IntraFlowDispatchPolicy`: The core interface that defines the contract for selecting an item from a flow's queue.
- `framework.FlowQueueAccessor`: A read-only interface that provides policies with safe access to queue state.
- `RequiredQueueCapabilities`: A mechanism for policies to declare their queue requirements (e.g., FIFO, priority-ordered), which are validated by the registry.
- A factory and registration system for discovering and instantiating policy plugins by name.
- A comprehensive conformance test suite to validate the contract for all policy plugins.
- A foundational `FCFS` (First-Come, First-Served) policy as the first reference implementation.

This work builds directly on the `SafeQueue` framework, enabling the development of sophisticated, policy-driven request prioritization and scheduling.

Author: LukeAVanDrie

pkg/epp/flowcontrol/framework/doc.go

@@ -20,8 +20,13 @@ limitations under the License.
 // to. By building on these interfaces, the Flow Control system can be extended and customized without modifying the
 // core controller logic.
 //
-// The primary interfaces defined here are:
-//   - `SafeQueue`: The contract for concurrent-safe queue implementations.
-//   - `ItemComparator`: The contract for policy-driven logic that defines the relative priority of items within a
+// The primary contracts are:
+//   - `SafeQueue`: An interface for concurrent-safe queue implementations.
+//   - `IntraFlowDispatchPolicy`: An interface for policies that decide which item to select from within a single flow's
 //     queue.
+//   - `ItemComparator`: An interface vended by policies to make their internal item-ordering logic explicit and
+//     available to other components.
+//
+// These components are linked by `QueueCapability`, which allows policies to declare their queue requirements (e.g.,
+// FIFO or priority-based ordering).
 package framework

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

@@ -18,6 +18,7 @@ package mocks
 
 import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 )
 
 // MockItemComparator provides a mock implementation of the `framework.ItemComparator` interface.
@@ -30,3 +31,35 @@ func (m *MockItemComparator) Func() framework.ItemComparatorFunc { return m.Func
 func (m *MockItemComparator) ScoreType() string                  { return m.ScoreTypeV }
 
 var _ framework.ItemComparator = &MockItemComparator{}
+
+// MockFlowQueueAccessor is a mock implementation of the `framework.FlowQueueAccessor` interface.
+type MockFlowQueueAccessor struct {
+	NameV         string
+	CapabilitiesV []framework.QueueCapability
+	LenV          int
+	ByteSizeV     uint64
+	PeekHeadV     types.QueueItemAccessor
+	PeekHeadErrV  error
+	PeekTailV     types.QueueItemAccessor
+	PeekTailErrV  error
+	FlowSpecV     types.FlowSpecification
+	ComparatorV   framework.ItemComparator
+}
+
+func (m *MockFlowQueueAccessor) Name() string                              { return m.NameV }
+func (m *MockFlowQueueAccessor) Capabilities() []framework.QueueCapability { return m.CapabilitiesV }
+func (m *MockFlowQueueAccessor) Len() int                                  { return m.LenV }
+func (m *MockFlowQueueAccessor) ByteSize() uint64                          { return m.ByteSizeV }
+
+func (m *MockFlowQueueAccessor) PeekHead() (types.QueueItemAccessor, error) {
+	return m.PeekHeadV, m.PeekHeadErrV
+}
+
+func (m *MockFlowQueueAccessor) PeekTail() (types.QueueItemAccessor, error) {
+	return m.PeekTailV, m.PeekTailErrV
+}
+
+func (m *MockFlowQueueAccessor) Comparator() framework.ItemComparator { return m.ComparatorV }
+func (m *MockFlowQueueAccessor) FlowSpec() types.FlowSpecification    { return m.FlowSpecV }
+
+var _ framework.FlowQueueAccessor = &MockFlowQueueAccessor{}

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

@@ -0,0 +1,63 @@
+# Flow Controller Intra-Flow Dispatch Policy Plugins
+
+This directory contains concrete implementations of the [`framework.IntraFlowDispatchPolicy`](../../../policies.go)
+interface. These policies are responsible for **temporal scheduling**: determining the order in which requests are
+selected for dispatch *from within a single flow's queue*.
+
+## Overview
+
+The `controller.FlowController` uses a two-tier policy system to manage requests. `framework.IntraFlowDispatchPolicy`
+plugins represent the first tier, making tactical decisions about the ordering of requests *within* a single logical
+flow (e.g., for a specific model or tenant).
+
+This contrasts with the `framework.InterFlowDispatchPolicy` (not yet implemented), which is responsible for
+**spatial fairness**: deciding *which flow's queue* gets the next opportunity to dispatch a request. The
+`framework.IntraFlowDispatchPolicy` only operates *after* the inter-flow policy has selected a specific queue.
+
+Key responsibilities and characteristics of an `IntraFlowDispatchPolicy`:
+
+1.  **Request Selection (`SelectItem`)**: The primary method, `SelectItem(queue framework.FlowQueueAccessor)`, inspects
+    the given flow's queue (via a read-only accessor) and decides which item, if any, should be dispatched next from
+    *that specific queue*.
+
+2.  **Priority Definition (`ItemComparator`)**:
+    - This policy type is unique because it defines the nature of priority for items *within its specific managed
+      queue*. It makes this logic explicit by vending a [`framework.ItemComparator`](../../../policies.go).
+    - The vended comparator defines the "less than" relationship between two items and exposes a `ScoreType()` string
+      (e.g., `"enqueue_time_ns_asc"`, `"slo_deadline_urgency"`) that gives a semantic meaning to the comparison.
+
+3.  **Queue Compatibility (`RequiredQueueCapabilities`)**: The policy specifies the capabilities its associated
+    [`framework.SafeQueue`](../../../queue.go) must support for it to function correctly. For example, a simple FCFS
+    policy would require `framework.CapabilityFIFO`, while a more complex, priority-based policy would require
+    `framework.CapabilityPriorityConfigurable`. The `ports.FlowRegistry` uses this information to pair policies with
+    compatible queues.
+
+The `IntraFlowDispatchPolicy` allows for fine-grained control over how individual requests within a single flow are
+serviced, enabling strategies like basic FCFS or more advanced schemes based on SLOs or deadlines.
+
+## Contributing a New `IntraFlowDispatchPolicy` 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.IntraFlowDispatchPolicy` 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 Conformance Test**
+    - Add a blank import for your new package to [`conformance_test.go`](./conformance_test.go). Your policy will then
+      be automatically included in the conformance suite, which validates the basic `framework.IntraFlowDispatchPolicy`
+      contract (e.g., correct initialization, handling of nil/empty queues).
+
+4.  **Add Policy-Specific Tests**
+    - The conformance 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 your `Comparator()` works as expected and that `SelectItem()`
+      correctly implements your desired selection logic for a non-empty queue.
+
+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/intraflow/dispatch/conformance_test.go

@@ -0,0 +1,79 @@
+/*
+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 dispatch_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+	frameworkmocks "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/mocks"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch"
+
+	_ "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch/fcfs"
+)
+
+// TestIntraFlowDispatchPolicyConformance is the main conformance test suite for `framework.IntraFlowDispatchPolicy`
+// implementations.
+// It iterates over all policy implementations registered via `dispatch.MustRegisterPolicy` and runs a series of
+// sub-tests to ensure they adhere to the `framework.IntraFlowDispatchPolicy` contract.
+func TestIntraFlowDispatchPolicyConformance(t *testing.T) {
+	t.Parallel()
+
+	for policyName, constructor := range dispatch.RegisteredPolicies {
+		t.Run(string(policyName), func(t *testing.T) {
+			t.Parallel()
+
+			policy, err := constructor()
+			require.NoError(t, err, "Policy constructor for %s failed", policyName)
+			require.NotNil(t, policy, "Constructor for %s should return a non-nil policy instance", policyName)
+
+			t.Run("Initialization", func(t *testing.T) {
+				t.Parallel()
+				assert.NotEmpty(t, policy.Name(), "Name() for %s should not be empty", policyName)
+
+				comp := policy.Comparator()
+				require.NotNil(t, comp, "Comparator() for %s should not return nil", policyName)
+				assert.NotNil(t, comp.Func(), "Comparator().Func() for %s should not be nil", policyName)
+				assert.NotEmpty(t, comp.ScoreType(), "Comparator().ScoreType() for %s should not be empty", policyName)
+
+				caps := policy.RequiredQueueCapabilities()
+				assert.NotNil(t, caps, "RequiredQueueCapabilities() for %s should not return a nil slice", policyName)
+			})
+
+			t.Run("SelectItemFromNilQueue", func(t *testing.T) {
+				t.Parallel()
+				item, err := policy.SelectItem(nil)
+				require.NoError(t, err, "SelectItem(nil) for %s should not return an error", policyName)
+				assert.Nil(t, item, "SelectItem(nil) for %s should return a nil item", policyName)
+			})
+
+			t.Run("SelectItemFromEmptyQueue", func(t *testing.T) {
+				t.Parallel()
+				mockQueue := &frameworkmocks.MockFlowQueueAccessor{
+					PeekHeadErrV: framework.ErrQueueEmpty,
+					LenV:         0,
+				}
+				item, err := policy.SelectItem(mockQueue)
+				require.NoError(t, err, "SelectItem from an empty queue for %s should not return an error", policyName)
+				assert.Nil(t, item, "SelectItem from an empty queue for %s should return a nil item", policyName)
+			})
+		})
+	}
+}

pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch/factory.go

@@ -0,0 +1,62 @@
+/*
+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 dispatch provides the factory and registration mechanism for all `IntraFlowDispatchPolicy` implementations.
+// It allows new policies to be added to the system and instantiated by name.
+package dispatch
+
+import (
+	"fmt"
+	"sync"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+)
+
+// RegisteredPolicyName is the unique name under which a policy is registered.
+type RegisteredPolicyName string
+
+// PolicyConstructor defines the function signature for creating a `framework.IntraFlowDispatchPolicy`.
+type PolicyConstructor func() (framework.IntraFlowDispatchPolicy, error)
+
+var (
+	// mu guards the registration map.
+	mu sync.RWMutex
+	// RegisteredPolicies stores the constructors for all registered policies.
+	RegisteredPolicies = make(map[RegisteredPolicyName]PolicyConstructor)
+)
+
+// MustRegisterPolicy registers a policy constructor, and panics if the name is already registered.
+// This is intended to be called from the `init()` function of a policy implementation.
+func MustRegisterPolicy(name RegisteredPolicyName, constructor PolicyConstructor) {
+	mu.Lock()
+	defer mu.Unlock()
+	if _, ok := RegisteredPolicies[name]; ok {
+		panic(fmt.Sprintf("IntraFlowDispatchPolicy already registered with name %q", name))
+	}
+	RegisteredPolicies[name] = constructor
+}
+
+// NewPolicyFromName creates a new `IntraFlowDispatchPolicy` given its registered name.
+// This is called by the `registry.FlowRegistry` when configuring a flow.
+func NewPolicyFromName(name RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
+	mu.RLock()
+	defer mu.RUnlock()
+	constructor, ok := RegisteredPolicies[name]
+	if !ok {
+		return nil, fmt.Errorf("no IntraFlowDispatchPolicy registered with name %q", name)
+	}
+	return constructor()
+}

pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch/fcfs/fcfs.go

@@ -0,0 +1,104 @@
+/*
+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 fcfs provides a First-Come, First-Served implementation of the `framework.IntraFlowDispatchPolicy`.
+package fcfs
+
+import (
+	"errors"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
+)
+
+// FCFSPolicyName is the name of the FCFS policy implementation.
+const FCFSPolicyName = "FCFS"
+
+func init() {
+	dispatch.MustRegisterPolicy(dispatch.RegisteredPolicyName(FCFSPolicyName),
+		func() (framework.IntraFlowDispatchPolicy, error) {
+			return newFCFS(), nil
+		})
+}
+
+// fcfs (First-Come, First-Served) implements the `framework.IntraFlowDispatchPolicy` interface.
+type fcfs struct {
+	comparator framework.ItemComparator
+}
+
+// newFCFS creates a new `fcfs` policy instance.
+func newFCFS() *fcfs {
+	return &fcfs{
+		comparator: &enqueueTimeComparator{},
+	}
+}
+
+// Name returns the name of the policy.
+func (p *fcfs) Name() string {
+	return FCFSPolicyName
+}
+
+// SelectItem selects the next item from the queue by peeking its head. This implementation relies on the queue being
+// ordered by dispatch preference, as indicated by its `RequiredQueueCapabilities`.
+func (p *fcfs) SelectItem(queue framework.FlowQueueAccessor) (types.QueueItemAccessor, error) {
+	if queue == nil {
+		return nil, nil
+	}
+	item, err := queue.PeekHead()
+	if errors.Is(err, framework.ErrQueueEmpty) {
+		return nil, nil
+	}
+	return item, err
+}
+
+// Comparator returns a `framework.ItemComparator` based on enqueue time.
+func (p *fcfs) Comparator() framework.ItemComparator {
+	return p.comparator
+}
+
+// RequiredQueueCapabilities specifies that this policy needs a queue that supports FIFO operations.
+func (p *fcfs) RequiredQueueCapabilities() []framework.QueueCapability {
+	return []framework.QueueCapability{framework.CapabilityFIFO}
+}
+
+// --- enqueueTimeComparator ---
+
+// enqueueTimeComparator implements `framework.ItemComparator` for FCFS logic.
+// It prioritizes items with earlier enqueue times.
+type enqueueTimeComparator struct{}
+
+// Func returns the comparison logic.
+// It returns true if item 'a' should be dispatched before item 'b'.
+func (c *enqueueTimeComparator) Func() framework.ItemComparatorFunc {
+	return func(a, b types.QueueItemAccessor) bool {
+		if a == nil && b == nil {
+			return false
+		}
+		if a == nil { // Treat nil as lowest priority
+			return false
+		}
+		if b == nil { // Treat non-nil 'a' as higher priority than nil 'b'
+			return true
+		}
+		return a.EnqueueTime().Before(b.EnqueueTime())
+	}
+}
+
+// ScoreType returns a string descriptor for the comparison logic.
+func (c *enqueueTimeComparator) ScoreType() string {
+	return string(framework.EnqueueTimePriorityScoreType)
+}