Repository: kubernetes-sigs/gateway-api-inference-extension

Issue:
Add support for PrepareRequestData and AdmitRequest plugins
This is to evolve the EPP datalayer based on the suggestions in [Evolving the EPP framework](https://docs.google.com/document/d/1EQwXL2pCuUyM1B917FUgP_8pFS3VF8F_bUfjy8IE7gM/edit?tab=t.vmaefhinvkl5#heading=h.s9mr6kynb3ls) doc.

Hints from issue comments:
@rahulgurnani: /assign@rahulgurnani: /assign@rahulgurnani: /milestone Data Layer Rework@rahulgurnani: /milestone Data Layer Rework


Solution patch:
diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go
index 24a59eb..d0490ca 100644
--- a/cmd/epp/runner/runner.go
+++ b/cmd/epp/runner/runner.go
@@ -139,8 +139,9 @@ var (
 	configFile = flag.String("config-file", runserver.DefaultConfigFile, "The path to the configuration file")
 	configText = flag.String("config-text", runserver.DefaultConfigText, "The configuration specified as text, in lieu of a file")
 
-	modelServerMetricsPort = flag.Int("model-server-metrics-port", 0, "Port to scrape metrics from pods. "+
-		"Default value will be set to the InferencePool.Spec.TargetPorts[0].Number if not set.")
+	modelServerMetricsPort = flag.Int("model-server-metrics-port", 0, "[DEPRECATED] Port to scrape metrics from pods. "+
+		"Default value will be set to the InferencePool.Spec.TargetPorts[0].Number if not set."+
+		"This option will be removed in the next release.")
 	modelServerMetricsPath                    = flag.String("model-server-metrics-path", "/metrics", "Path to scrape metrics from pods")
 	modelServerMetricsScheme                  = flag.String("model-server-metrics-scheme", "http", "Scheme to scrape metrics from pods")
 	modelServerMetricsHttpsInsecureSkipVerify = flag.Bool("model-server-metrics-https-insecure-skip-verify", true, "When using 'https' scheme for 'model-server-metrics-scheme', configure 'InsecureSkipVerify' (default to true)")
@@ -198,6 +199,8 @@ func (r *Runner) Run(ctx context.Context) error {
 	flag.Parse()
 	initLogging(&opts)
 
+	r.deprecatedFlagsHandler(setupLog)
+
 	if *tracing {
 		err := common.InitTracing(ctx, setupLog)
 		if err != nil {
@@ -478,6 +481,10 @@ func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *conf
 
 	// Add requestControl plugins
 	r.requestControlConfig.AddPlugins(handle.GetAllPlugins()...)
+	// Sort prepare data plugins in DAG order (topological sort). Also check prepare data plugins for cycles.
+	if r.requestControlConfig.PrepareDataPluginGraph() != nil {
+		return nil, errors.New("failed to load the configuration - prepare data plugins have cyclic dependencies")
+	}
 
 	// Handler deprecated configuration options
 	r.deprecatedConfigurationHelper(cfg, logger)
@@ -486,6 +493,14 @@ func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *conf
 	return cfg, nil
 }
 
+func (r *Runner) deprecatedFlagsHandler(logger logr.Logger) {
+	flag.Visit(func(f *flag.Flag) {
+		if f.Name == "model-server-metrics-port" { // future: use  map/set to store deprecated flags (and replacements?)
+			logger.Info("deprecated option will be removed in the next release.", "option", f.Name)
+		}
+	})
+}
+
 func (r *Runner) deprecatedConfigurationHelper(cfg *config.Config, logger logr.Logger) {
 	// Handle deprecated environment variable based feature flags
 
diff --git a/config/manifests/vllm/sim-deployment.yaml b/config/manifests/vllm/sim-deployment.yaml
index 2415c10..be2388c 100644
--- a/config/manifests/vllm/sim-deployment.yaml
+++ b/config/manifests/vllm/sim-deployment.yaml
@@ -14,7 +14,7 @@ spec:
     spec:
       containers:
       - name: vllm-sim
-        image: ghcr.io/llm-d/llm-d-inference-sim:v0.5.0
+        image: ghcr.io/llm-d/llm-d-inference-sim:v0.6.1
         imagePullPolicy: Always
         args:
         - --model
diff --git a/conformance/resources/base.yaml b/conformance/resources/base.yaml
index 3b4933c..87de80a 100644
--- a/conformance/resources/base.yaml
+++ b/conformance/resources/base.yaml
@@ -200,7 +200,7 @@ spec:
       terminationGracePeriodSeconds: 130
       containers:
       - name: epp
-        image: us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-inference-extension/epp:v20251105-cbb8928
+        image: us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-inference-extension/epp:v20251119-2aaf2a6
         imagePullPolicy: Always
         args:
         - --pool-name
@@ -298,7 +298,7 @@ spec:
       terminationGracePeriodSeconds: 130
       containers:
       - name: epp
-        image: us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-inference-extension/epp:v20251105-cbb8928
+        image: us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-inference-extension/epp:v20251119-2aaf2a6
         imagePullPolicy: Always
         args:
         - --pool-name
@@ -507,7 +507,7 @@ spec:
       terminationGracePeriodSeconds: 130
       containers:
       - name: epp
-        image: us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-inference-extension/epp:v20251105-cbb8928
+        image: us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-inference-extension/epp:v20251119-2aaf2a6
         imagePullPolicy: Always
         args:
         - --pool-name
@@ -560,6 +560,7 @@ data:
     kind: EndpointPickerConfig
     plugins:
     - type: header-based-testing-filter
+    - type: destination-endpoint-served-verifier
     schedulingProfiles:
     - name: conformance-profile
       plugins:
diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go
index 2ab2e98..d4f14b9 100644
--- a/pkg/epp/datastore/datastore.go
+++ b/pkg/epp/datastore/datastore.go
@@ -68,6 +68,8 @@ type Datastore interface {
 	Clear()
 }
 
+// NewDatastore creates a new data store.
+// TODO: modelServerMetricsPort is being deprecated
 func NewDatastore(parentCtx context.Context, epFactory datalayer.EndpointFactory, modelServerMetricsPort int32, opts ...DatastoreOption) Datastore {
 	// Initialize with defaults
 	store := &datastore{
@@ -100,7 +102,7 @@ type datastore struct {
 	pods *sync.Map
 	// modelServerMetricsPort metrics port from EPP command line argument
 	// used only if there is only one inference engine per pod
-	modelServerMetricsPort int32
+	modelServerMetricsPort int32 // TODO: deprecating
 	epf                    datalayer.EndpointFactory
 }
 
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/README.md b/pkg/epp/flowcontrol/framework/plugins/interflow/README.md
similarity index 100%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/README.md
rename to pkg/epp/flowcontrol/framework/plugins/interflow/README.md
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead.go b/pkg/epp/flowcontrol/framework/plugins/interflow/besthead.go
similarity index 93%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead.go
rename to pkg/epp/flowcontrol/framework/plugins/interflow/besthead.go
index ae4cbd6..3bfc87f 100644
--- a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead.go
+++ b/pkg/epp/flowcontrol/framework/plugins/interflow/besthead.go
@@ -16,13 +16,12 @@ 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
+package interflow
 
 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"
 )
 
@@ -30,7 +29,7 @@ import (
 const BestHeadPolicyName = "BestHead"
 
 func init() {
-	dispatch.MustRegisterPolicy(dispatch.RegisteredPolicyName(BestHeadPolicyName),
+	MustRegisterPolicy(RegisteredPolicyName(BestHeadPolicyName),
 		func() (framework.InterFlowDispatchPolicy, error) {
 			return newBestHead(), nil
 		})
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/factory.go b/pkg/epp/flowcontrol/framework/plugins/interflow/factory.go
similarity index 99%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/factory.go
rename to pkg/epp/flowcontrol/framework/plugins/interflow/factory.go
index 4cebd6c..0f5bf15 100644
--- a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/factory.go
+++ b/pkg/epp/flowcontrol/framework/plugins/interflow/factory.go
@@ -17,7 +17,7 @@ limitations under the License.
 // Package dispatch provides the factory and registration mechanism for all `framework.InterFlowDispatchPolicy`
 // implementations.
 // It allows new policies to be added to the system and instantiated by name.
-package dispatch
+package interflow
 
 import (
 	"fmt"
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin/roundrobin.go b/pkg/epp/flowcontrol/framework/plugins/interflow/roundrobin.go
similarity index 94%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin/roundrobin.go
rename to pkg/epp/flowcontrol/framework/plugins/interflow/roundrobin.go
index c71b801..b4d6d54 100644
--- a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin/roundrobin.go
+++ b/pkg/epp/flowcontrol/framework/plugins/interflow/roundrobin.go
@@ -16,14 +16,13 @@ limitations under the License.
 
 // Package roundrobin provides a `framework.InterFlowDispatchPolicy` that selects a queue from a priority band using a
 // simple round-robin strategy.
-package roundrobin
+package interflow
 
 import (
 	"slices"
 	"sync"
 
 	"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"
 )
 
@@ -31,7 +30,7 @@ import (
 const RoundRobinPolicyName = "RoundRobin"
 
 func init() {
-	dispatch.MustRegisterPolicy(dispatch.RegisteredPolicyName(RoundRobinPolicyName),
+	MustRegisterPolicy(RegisteredPolicyName(RoundRobinPolicyName),
 		func() (framework.InterFlowDispatchPolicy, error) {
 			return newRoundRobin(), nil
 		})
@@ -104,7 +103,7 @@ func (r *iterator) selectNextQueue(band framework.PriorityBandAccessor) framewor
 	}
 
 	numFlows := len(keys)
-	for i := 0; i < numFlows; i++ {
+	for i := range numFlows {
 		currentIdx := (startIndex + i) % numFlows
 		currentKey := keys[currentIdx]
 		queue := band.Queue(currentKey.ID)
diff --git a/pkg/epp/flowcontrol/registry/config.go b/pkg/epp/flowcontrol/registry/config.go
index 2175782..6ded14d 100644
--- a/pkg/epp/flowcontrol/registry/config.go
+++ b/pkg/epp/flowcontrol/registry/config.go
@@ -23,8 +23,7 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
-	inter "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/framework/plugins/policies/interflow/dispatch/besthead"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/interflow"
 	intra "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"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
@@ -41,7 +40,7 @@ const (
 	// defaultIntraFlowDispatchPolicy is the default policy for selecting items within a single flow's queue.
 	defaultIntraFlowDispatchPolicy intra.RegisteredPolicyName = fcfs.FCFSPolicyName
 	// defaultInterFlowDispatchPolicy is the default policy for selecting which flow's queue to service next.
-	defaultInterFlowDispatchPolicy inter.RegisteredPolicyName = besthead.BestHeadPolicyName
+	defaultInterFlowDispatchPolicy interflow.RegisteredPolicyName = interflow.BestHeadPolicyName
 	// defaultQueue is the default queue implementation for flows.
 	defaultQueue queue.RegisteredQueueName = queue.ListQueueName
 	// defaultInitialShardCount is the default number of parallel shards to create when the registry is initialized.
@@ -128,7 +127,7 @@ type PriorityBandConfig struct {
 	// InterFlowDispatchPolicy specifies the name of the policy used to select which flow's queue to service next from
 	// this band.
 	// Optional: Defaults to `defaultInterFlowDispatchPolicy` ("BestHead").
-	InterFlowDispatchPolicy inter.RegisteredPolicyName
+	InterFlowDispatchPolicy interflow.RegisteredPolicyName
 
 	// Queue specifies the default name of the `framework.SafeQueue` implementation for flow queues in this band.
 	// Optional: Defaults to `defaultQueue` ("ListQueue").
@@ -167,7 +166,7 @@ type ShardPriorityBandConfig struct {
 	// IntraFlowDispatchPolicy is the name of the policy for dispatch within a flow's queue.
 	IntraFlowDispatchPolicy intra.RegisteredPolicyName
 	// InterFlowDispatchPolicy is the name of the policy for dispatch between flow queues.
-	InterFlowDispatchPolicy inter.RegisteredPolicyName
+	InterFlowDispatchPolicy interflow.RegisteredPolicyName
 	// Queue is the name of the queue implementation to use.
 	Queue queue.RegisteredQueueName
 	// MaxBytes is this shard's partitioned portion of this band's global capacity limit.
@@ -209,7 +208,7 @@ func (c *Config) ValidateAndApplyDefaults() (*Config, error) {
 
 	// Ensure the DI factories are initialized for production use if `NewConfig` was called without options.
 	if cfg.interFlowDispatchPolicyFactory == nil {
-		cfg.interFlowDispatchPolicyFactory = inter.NewPolicyFromName
+		cfg.interFlowDispatchPolicyFactory = interflow.NewPolicyFromName
 	}
 	if cfg.intraFlowDispatchPolicyFactory == nil {
 		cfg.intraFlowDispatchPolicyFactory = intra.NewPolicyFromName
@@ -362,9 +361,9 @@ type configOption func(*Config)
 
 // interFlowDispatchPolicyFactory defines the signature for a function that creates an
 // `framework.InterFlowDispatchPolicy` instance from its registered name.
-// It serves as an abstraction over the concrete `inter.NewPolicyFromName` factory, enabling dependency injection for
+// It serves as an abstraction over the concrete `interflow.NewPolicyFromName` factory, enabling dependency injection for
 // testing validation logic.
-type interFlowDispatchPolicyFactory func(name inter.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error)
+type interFlowDispatchPolicyFactory func(name interflow.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error)
 
 // intraFlowDispatchPolicyFactory defines the signature for a function that creates an
 // `framework.IntraFlowDispatchPolicy` instance from its registered name.
diff --git a/pkg/epp/requestcontrol/dag.go b/pkg/epp/requestcontrol/dag.go
new file mode 100644
index 0000000..8ae7904
--- /dev/null
+++ b/pkg/epp/requestcontrol/dag.go
@@ -0,0 +1,127 @@
+/*
+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 requestcontrol
+
+import (
+	"errors"
+	"slices"
+)
+
+// buildDAG builds a dependency graph among data preparation plugins based on their
+// produced and consumed data keys.
+func buildDAG(plugins []PrepareDataPlugin) map[string][]string {
+	dag := make(map[string][]string)
+	for _, plugin := range plugins {
+		dag[plugin.TypedName().String()] = []string{}
+	}
+	// Create dependency graph as a DAG.
+	for i := range plugins {
+		for j := range plugins {
+			if i == j {
+				continue
+			}
+			// Check whether plugin[i] produces something consumed by plugin[j]. In that case, j depends on i.
+			if plugins[i].Produces() != nil && plugins[j].Consumes() != nil {
+				// For all the keys produced by plugin i, check if plugin j consumes any of them.
+				// If yes, then j depends on i.
+				for producedKey := range plugins[i].Produces() {
+					// If plugin j consumes the produced key, then j depends on i. We can break after the first match.
+					if _, ok := plugins[j].Consumes()[producedKey]; ok {
+						iPluginName := plugins[i].TypedName().String()
+						jPluginName := plugins[j].TypedName().String()
+						dag[jPluginName] = append(dag[jPluginName], iPluginName)
+						break
+					}
+				}
+			}
+		}
+	}
+	return dag
+}
+
+// sortPlugins builds the dependency graph and returns the plugins ordered in topological order.
+// If there is a cycle, it returns an error.
+func sortPlugins(dag map[string][]string, plugins []PrepareDataPlugin) ([]PrepareDataPlugin, error) {
+	nameToPlugin := map[string]PrepareDataPlugin{}
+	for _, plugin := range plugins {
+		nameToPlugin[plugin.TypedName().String()] = plugin
+	}
+	sortedPlugins, err := topologicalSort(dag)
+	if err != nil {
+		return nil, err
+	}
+	orderedPlugins := []PrepareDataPlugin{}
+	for _, pluginName := range sortedPlugins {
+		orderedPlugins = append(orderedPlugins, nameToPlugin[pluginName])
+	}
+
+	return orderedPlugins, err
+}
+
+// TopologicalSort performs Kahn's Algorithm on a DAG.
+// It returns the sorted order or an error if a cycle is detected.
+func topologicalSort(graph map[string][]string) ([]string, error) {
+	// 1. Initialize in-degree map
+	inDegree := make(map[string]int)
+
+	// Ensure all nodes are present in the inDegree map, even those with no dependencies
+	for u, neighbors := range graph {
+		if _, ok := inDegree[u]; !ok {
+			inDegree[u] = 0
+		}
+		for _, v := range neighbors {
+			inDegree[v]++ // Increment in-degree for the destination node
+		}
+	}
+
+	// 2. Initialize the queue with nodes having 0 in-degree
+	var queue []string
+	for node, degree := range inDegree {
+		if degree == 0 {
+			queue = append(queue, node)
+		}
+	}
+
+	var result []string
+
+	// 3. Process the queue
+	for len(queue) > 0 {
+		// Dequeue
+		u := queue[0]
+		queue = queue[1:]
+
+		result = append(result, u)
+
+		// Decrease in-degree of neighbors
+		if neighbors, ok := graph[u]; ok {
+			for _, v := range neighbors {
+				inDegree[v]--
+				if inDegree[v] == 0 {
+					queue = append(queue, v)
+				}
+			}
+		}
+	}
+
+	// 4. Check for cycles
+	// If the result size != total nodes, there is a cycle
+	if len(result) != len(inDegree) {
+		return nil, errors.New("cycle detected: graph is not a DAG")
+	}
+	slices.Reverse(result)
+	return result, nil
+}
diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go
index c4f4f1c..7a77c4e 100644
--- a/pkg/epp/requestcontrol/director.go
+++ b/pkg/epp/requestcontrol/director.go
@@ -351,13 +351,9 @@ func (d *Director) runPreRequestPlugins(ctx context.Context, request *scheduling
 	}
 }
 
-// TODO: Execute plugins in parallel once DAG execution is supported.
-// runPrepareDataPlugins executes PrepareDataPlugins sequentially.
 func (d *Director) runPrepareDataPlugins(ctx context.Context,
 	request *schedulingtypes.LLMRequest, pods []schedulingtypes.Pod) error {
-	return prepareDataPluginsWithTimeout(
-		prepareDataTimeout, d.requestControlPlugins.prepareDataPlugins, ctx, request, pods)
-
+	return prepareDataPluginsWithTimeout(prepareDataTimeout, d.requestControlPlugins.prepareDataPlugins, ctx, request, pods)
 }
 
 func (d *Director) runAdmissionPlugins(ctx context.Context,
diff --git a/pkg/epp/requestcontrol/plugin_executor.go b/pkg/epp/requestcontrol/plugin_executor.go
index be8d4ba..7771c4f 100644
--- a/pkg/epp/requestcontrol/plugin_executor.go
+++ b/pkg/epp/requestcontrol/plugin_executor.go
@@ -24,20 +24,24 @@ import (
 	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
+// executePluginsAsDAG executes PrepareData plugins as a DAG based on their dependencies asynchronously.
+// So, a plugin is executed only after all its dependencies have been executed.
+// If there is a cycle or any plugin fails with error, it returns an error.
+func executePluginsAsDAG(plugins []PrepareDataPlugin, ctx context.Context, request *schedulingtypes.LLMRequest, pods []schedulingtypes.Pod) error {
+	for _, plugin := range plugins {
+		if err := plugin.PrepareRequestData(ctx, request, pods); err != nil {
+			return errors.New("prepare data plugin " + plugin.TypedName().String() + " failed: " + err.Error())
+		}
+	}
+	return nil
+}
+
 // prepareDataPluginsWithTimeout executes the PrepareRequestData plugins with retries and timeout.
 func prepareDataPluginsWithTimeout(timeout time.Duration, plugins []PrepareDataPlugin,
 	ctx context.Context, request *schedulingtypes.LLMRequest, pods []schedulingtypes.Pod) error {
 	errCh := make(chan error, 1)
-	// Execute plugins sequentially in a separate goroutine
 	go func() {
-		for _, plugin := range plugins {
-			err := plugin.PrepareRequestData(ctx, request, pods)
-			if err != nil {
-				errCh <- errors.New("prepare data plugin " + plugin.TypedName().String() + " failed: " + err.Error())
-				return
-			}
-		}
-		errCh <- nil
+		errCh <- executePluginsAsDAG(plugins, ctx, request, pods)
 	}()
 
 	select {
diff --git a/pkg/epp/requestcontrol/request_control_config.go b/pkg/epp/requestcontrol/request_control_config.go
index 9701be9..8f08ac1 100644
--- a/pkg/epp/requestcontrol/request_control_config.go
+++ b/pkg/epp/requestcontrol/request_control_config.go
@@ -104,3 +104,16 @@ func (c *Config) AddPlugins(pluginObjects ...plugins.Plugin) {
 		}
 	}
 }
+
+// PrepareDataPluginGraph creates data dependency graph and sorts the plugins in topological order.
+// If a cycle is detected, it returns an error.
+func (c *Config) PrepareDataPluginGraph() error {
+	dag := buildDAG(c.prepareDataPlugins)
+	plugins, err := sortPlugins(dag, c.prepareDataPlugins)
+	if err != nil {
+		return err
+	}
+	c.prepareDataPlugins = plugins
+
+	return nil
+}



Test patch:
diff --git a/conformance/tests/gateway_destination_endpoint_served.go b/conformance/tests/gateway_destination_endpoint_served.go
new file mode 100644
index 0000000..862abf4
--- /dev/null
+++ b/conformance/tests/gateway_destination_endpoint_served.go
@@ -0,0 +1,109 @@
+/*
+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 tests
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/types"
+	gwhttp "sigs.k8s.io/gateway-api/conformance/utils/http"
+	"sigs.k8s.io/gateway-api/conformance/utils/suite"
+	"sigs.k8s.io/gateway-api/pkg/features"
+
+	"sigs.k8s.io/gateway-api-inference-extension/conformance/resources"
+	k8sutils "sigs.k8s.io/gateway-api-inference-extension/conformance/utils/kubernetes"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/requestcontrol/plugins/test"
+	testscheduling "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/test"
+)
+
+func init() {
+	ConformanceTests = append(ConformanceTests, GatewayDestinationEndpointServed)
+}
+
+var GatewayDestinationEndpointServed = suite.ConformanceTest{
+	ShortName:   "GatewayDestinationEndpointServed",
+	Description: "A conformance test to verify that the gateway correctly reports the endpoint that served the request.",
+	Manifests:   []string{"tests/gateway_destination_endpoint_served.yaml"},
+	Features: []features.FeatureName{
+		features.FeatureName("SupportInferencePool"),
+		features.SupportGateway,
+	},
+	Test: func(t *testing.T, s *suite.ConformanceTestSuite) {
+		const (
+			hostname = "primary.example.com"
+			path     = "/destination-endpoint-served-test"
+		)
+
+		httpRouteNN := types.NamespacedName{Name: "httproute-for-destination-endpoint-served", Namespace: resources.AppBackendNamespace}
+		gatewayNN := resources.PrimaryGatewayNN
+		poolNN := resources.PrimaryInferencePoolNN
+		backendPodLabels := map[string]string{"app": resources.PrimaryModelServerAppLabel}
+
+		t.Log("Verifying HTTPRoute and InferencePool are accepted and the Gateway has an address.")
+		k8sutils.HTTPRouteMustBeAcceptedAndResolved(t, s.Client, s.TimeoutConfig, httpRouteNN, gatewayNN)
+		k8sutils.InferencePoolMustBeAcceptedByParent(t, s.Client, poolNN, gatewayNN)
+		gwAddr := k8sutils.GetGatewayEndpoint(t, s.Client, s.TimeoutConfig, gatewayNN)
+
+		t.Logf("Fetching backend pods with labels: %v", backendPodLabels)
+		pods, err := k8sutils.GetPodsWithLabel(t, s.Client, resources.AppBackendNamespace, backendPodLabels, s.TimeoutConfig)
+		require.NoError(t, err, "Failed to get backend pods")
+		require.Len(t, pods, resources.ModelServerPodReplicas, "Expected to find %d backend pods, but found %d.", resources.ModelServerPodReplicas, len(pods))
+
+		podIPs := make([]string, len(pods))
+		podNames := make([]string, len(pods))
+		for i, pod := range pods {
+			podIPs[i] = pod.Status.PodIP
+			podNames[i] = pod.Name
+		}
+
+		requestBody := `{
+            "model": "conformance-fake-model",
+            "prompt": "Write as if you were a critic: San Francisco"
+        }`
+		t.Run("Request is served by the selected backend pod", func(t *testing.T) {
+			for i := 0; i < len(pods); i++ {
+				gwhttp.MakeRequestAndExpectEventuallyConsistentResponse(
+					t,
+					s.RoundTripper,
+					s.TimeoutConfig,
+					gwAddr,
+					gwhttp.ExpectedResponse{
+						Request: gwhttp.Request{
+							Host:   hostname,
+							Path:   path,
+							Method: http.MethodPost,
+							Body:   requestBody,
+							Headers: map[string]string{
+								testscheduling.HeaderTestEppEndPointSelectionKey: podIPs[i],
+							},
+						},
+						Response: gwhttp.Response{
+							StatusCodes: []int{http.StatusOK},
+							Headers: map[string]string{
+								test.ConformanceTestResultHeader: podIPs[i] + ":3000", // The echo server's port is 3000.
+							},
+						},
+						Backend:   podNames[i],
+						Namespace: resources.AppBackendNamespace,
+					},
+				)
+			}
+		})
+	},
+}
diff --git a/conformance/tests/gateway_destination_endpoint_served.yaml b/conformance/tests/gateway_destination_endpoint_served.yaml
new file mode 100644
index 0000000..0b15da1
--- /dev/null
+++ b/conformance/tests/gateway_destination_endpoint_served.yaml
@@ -0,0 +1,23 @@
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: httproute-for-destination-endpoint-served
+  namespace: inference-conformance-app-backend
+spec:
+  parentRefs:
+  - group: gateway.networking.k8s.io
+    kind: Gateway
+    name: conformance-primary
+    namespace: inference-conformance-infra
+    sectionName: http
+  hostnames:
+  - "primary.example.com"
+  rules:
+  - backendRefs:
+    - group: inference.networking.k8s.io
+      kind: InferencePool
+      name: primary-inference-pool
+    matches:
+    - path:
+        type: PathPrefix
+        value: /destination-endpoint-served-test
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead_test.go b/pkg/epp/flowcontrol/framework/plugins/interflow/besthead_test.go
similarity index 97%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead_test.go
rename to pkg/epp/flowcontrol/framework/plugins/interflow/besthead_test.go
index 2885603..efb4463 100644
--- a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead/besthead_test.go
+++ b/pkg/epp/flowcontrol/framework/plugins/interflow/besthead_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package besthead
+package interflow
 
 import (
 	"testing"
@@ -35,11 +35,6 @@ const (
 	commonScoreType = "enqueue_time_ns_asc"
 )
 
-var (
-	flow1Key = types.FlowKey{ID: flow1ID, Priority: 0}
-	flow2Key = types.FlowKey{ID: flow2ID, Priority: 0}
-)
-
 // enqueueTimeComparatorFunc is a test utility. Lower enqueue time is better.
 func enqueueTimeComparatorFunc(a, b types.QueueItemAccessor) bool {
 	return a.EnqueueTime().Before(b.EnqueueTime())
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/functional_test.go b/pkg/epp/flowcontrol/framework/plugins/interflow/functional_test.go
similarity index 90%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/functional_test.go
rename to pkg/epp/flowcontrol/framework/plugins/interflow/functional_test.go
index 3dba8b7..3513c98 100644
--- a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/functional_test.go
+++ b/pkg/epp/flowcontrol/framework/plugins/interflow/functional_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package dispatch_test
+package interflow
 
 import (
 	"testing"
@@ -24,9 +24,6 @@ import (
 
 	"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/interflow/dispatch"
-	_ "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/besthead"
-	_ "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 )
 
@@ -37,7 +34,7 @@ import (
 func TestInterFlowDispatchPolicyConformance(t *testing.T) {
 	t.Parallel()
 
-	for policyName, constructor := range dispatch.RegisteredPolicies {
+	for policyName, constructor := range RegisteredPolicies {
 		t.Run(string(policyName), func(t *testing.T) {
 			t.Parallel()
 
diff --git a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin/roundrobin_test.go b/pkg/epp/flowcontrol/framework/plugins/interflow/roundrobin_test.go
similarity index 99%
rename from pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin/roundrobin_test.go
rename to pkg/epp/flowcontrol/framework/plugins/interflow/roundrobin_test.go
index dd4a026..9f6c9e3 100644
--- a/pkg/epp/flowcontrol/framework/plugins/policies/interflow/dispatch/roundrobin/roundrobin_test.go
+++ b/pkg/epp/flowcontrol/framework/plugins/interflow/roundrobin_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package roundrobin
+package interflow
 
 import (
 	"fmt"
diff --git a/pkg/epp/flowcontrol/registry/config_test.go b/pkg/epp/flowcontrol/registry/config_test.go
index b95f013..05d8e57 100644
--- a/pkg/epp/flowcontrol/registry/config_test.go
+++ b/pkg/epp/flowcontrol/registry/config_test.go
@@ -28,7 +28,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
 	"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/interflow/dispatch/besthead"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/interflow"
 	intra "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"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
@@ -82,7 +82,7 @@ func TestConfig_ValidateAndApplyDefaults(t *testing.T) {
 					Priority:                1,
 					PriorityName:            "Critical",
 					IntraFlowDispatchPolicy: fcfs.FCFSPolicyName,
-					InterFlowDispatchPolicy: besthead.BestHeadPolicyName,
+					InterFlowDispatchPolicy: interflow.BestHeadPolicyName,
 					Queue:                   queue.ListQueueName,
 					MaxBytes:                500,
 				}},
diff --git a/pkg/epp/flowcontrol/registry/registry_test.go b/pkg/epp/flowcontrol/registry/registry_test.go
index b5bc322..494bce0 100644
--- a/pkg/epp/flowcontrol/registry/registry_test.go
+++ b/pkg/epp/flowcontrol/registry/registry_test.go
@@ -31,7 +31,7 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
-	inter "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/framework/plugins/interflow"
 	intra "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"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types/mocks"
@@ -139,7 +139,7 @@ func TestFlowRegistry_New(t *testing.T) {
 		t.Parallel()
 		config, err := newConfig(
 			Config{PriorityBands: []PriorityBandConfig{{Priority: highPriority, PriorityName: "A"}}},
-			withInterFlowDispatchPolicyFactory(func(inter.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error) {
+			withInterFlowDispatchPolicyFactory(func(interflow.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error) {
 				return nil, errors.New("injected factory failure")
 			}),
 		)
diff --git a/pkg/epp/flowcontrol/registry/shard_test.go b/pkg/epp/flowcontrol/registry/shard_test.go
index 23bf813..c58b71c 100644
--- a/pkg/epp/flowcontrol/registry/shard_test.go
+++ b/pkg/epp/flowcontrol/registry/shard_test.go
@@ -28,7 +28,7 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
-	inter "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/framework/plugins/interflow"
 	intra "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/queue"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
@@ -73,7 +73,7 @@ func newShardTestHarness(t *testing.T) *shardTestHarness {
 		"test-shard-1",
 		shardConfig, logr.Discard(),
 		statsPropagator.propagate,
-		inter.NewPolicyFromName,
+		interflow.NewPolicyFromName,
 	)
 	require.NoError(t, err, "Test setup: newShard should not return an error with valid configuration")
 
@@ -149,7 +149,7 @@ func TestShard_New(t *testing.T) {
 		shardConfig, _ := newConfig(Config{PriorityBands: []PriorityBandConfig{
 			{Priority: highPriority, PriorityName: "High"},
 		}})
-		failingFactory := func(inter.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error) {
+		failingFactory := func(interflow.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error) {
 			return nil, errors.New("policy not found")
 		}
 		_, err := newShard("test-shard-1", shardConfig.partition(0, 1), logr.Discard(), nil, failingFactory)
diff --git a/pkg/epp/requestcontrol/dag_test.go b/pkg/epp/requestcontrol/dag_test.go
new file mode 100644
index 0000000..e2227ea
--- /dev/null
+++ b/pkg/epp/requestcontrol/dag_test.go
@@ -0,0 +1,159 @@
+/*
+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 requestcontrol
+
+import (
+	"context"
+	"maps"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+type mockPrepareRequestDataP struct {
+	name     string
+	produces map[string]any
+	consumes map[string]any
+}
+
+func (m *mockPrepareRequestDataP) TypedName() plugins.TypedName {
+	return plugins.TypedName{Name: m.name, Type: "mock"}
+}
+
+func (m *mockPrepareRequestDataP) Produces() map[string]any {
+	return m.produces
+}
+
+func (m *mockPrepareRequestDataP) Consumes() map[string]any {
+	return m.consumes
+}
+
+func (m *mockPrepareRequestDataP) PrepareRequestData(ctx context.Context, request *types.LLMRequest, pods []types.Pod) error {
+	pods[0].Put(mockProducedDataKey, mockProducedDataType{value: 42})
+	return nil
+}
+
+func TestPrepareDataGraph(t *testing.T) {
+	pluginA := &mockPrepareRequestDataP{name: "A", produces: map[string]any{"keyA": nil}}
+	pluginB := &mockPrepareRequestDataP{name: "B", consumes: map[string]any{"keyA": nil}, produces: map[string]any{"keyB": nil}}
+	pluginC := &mockPrepareRequestDataP{name: "C", consumes: map[string]any{"keyB": nil}}
+	pluginD := &mockPrepareRequestDataP{name: "D", consumes: map[string]any{"keyA": nil}}
+	pluginE := &mockPrepareRequestDataP{name: "E"} // No dependencies
+
+	// Cycle plugins
+	pluginX := &mockPrepareRequestDataP{name: "X", produces: map[string]any{"keyX": nil}, consumes: map[string]any{"keyY": nil}}
+	pluginY := &mockPrepareRequestDataP{name: "Y", produces: map[string]any{"keyY": nil}, consumes: map[string]any{"keyX": nil}}
+
+	testCases := []struct {
+		name        string
+		plugins     []PrepareDataPlugin
+		expectedDAG map[string][]string
+		expectError bool
+	}{
+		{
+			name:        "No plugins",
+			plugins:     []PrepareDataPlugin{},
+			expectedDAG: map[string][]string{},
+			expectError: false,
+		},
+		{
+			name:    "Plugins with no dependencies",
+			plugins: []PrepareDataPlugin{pluginA, pluginE},
+			expectedDAG: map[string][]string{
+				"A/mock": {},
+				"E/mock": {},
+			},
+			expectError: false,
+		},
+		{
+			name:    "Simple linear dependency (C -> B -> A)",
+			plugins: []PrepareDataPlugin{pluginA, pluginB, pluginC},
+			expectedDAG: map[string][]string{
+				"A/mock": {},
+				"B/mock": {"A/mock"},
+				"C/mock": {"B/mock"},
+			},
+			expectError: false,
+		},
+		{
+			name:    "DAG with multiple dependencies (B -> A, D -> A, E independent)",
+			plugins: []PrepareDataPlugin{pluginA, pluginB, pluginD, pluginE},
+			expectedDAG: map[string][]string{
+				"A/mock": {},
+				"B/mock": {"A/mock"},
+				"D/mock": {"A/mock"},
+				"E/mock": {},
+			},
+			expectError: false,
+		},
+		{
+			name:        "Graph with a cycle (X -> Y, Y -> X)",
+			plugins:     []PrepareDataPlugin{pluginX, pluginY},
+			expectedDAG: nil,
+			expectError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			dag := buildDAG(tc.plugins)
+			orderedPlugins, err := sortPlugins(dag, tc.plugins)
+
+			if tc.expectError {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "cycle detected")
+				return
+			}
+			assert.NoError(t, err)
+
+			// Normalize the slices in the maps for consistent comparison
+			normalizedDAG := make(map[string][]string)
+			maps.Copy(normalizedDAG, dag)
+			normalizedExpectedDAG := make(map[string][]string)
+			for k, v := range tc.expectedDAG {
+				normalizedExpectedDAG[k] = v
+			}
+
+			if diff := cmp.Diff(normalizedExpectedDAG, normalizedDAG); diff != "" {
+				t.Errorf("prepareDataGraph() mismatch (-want +got):\n%s", diff)
+			}
+
+			orderedPluginNames := make([]string, len(orderedPlugins))
+			for i, p := range orderedPlugins {
+				orderedPluginNames[i] = p.TypedName().String()
+			}
+			assertTopologicalOrder(t, dag, orderedPlugins)
+		})
+	}
+}
+
+func assertTopologicalOrder(t *testing.T, dag map[string][]string, ordered []PrepareDataPlugin) {
+	t.Helper()
+	positions := make(map[string]int)
+	for i, p := range ordered {
+		positions[p.TypedName().String()] = i
+	}
+
+	for node, dependencies := range dag {
+		for _, dep := range dependencies {
+			assert.Less(t, positions[dep], positions[node], "Dependency %s should come before %s", dep, node)
+		}
+	}
+}
diff --git a/pkg/epp/requestcontrol/plugin_executor_test.go b/pkg/epp/requestcontrol/plugin_executor_test.go
index b37264f..1825f7a 100644
--- a/pkg/epp/requestcontrol/plugin_executor_test.go
+++ b/pkg/epp/requestcontrol/plugin_executor_test.go
@@ -19,6 +19,7 @@ package requestcontrol
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -157,3 +158,138 @@ func TestPrepareDataPluginsWithTimeout(t *testing.T) {
 		})
 	}
 }
+
+type dagTestPlugin struct {
+	mockPrepareRequestDataPlugin
+	produces map[string]any
+	consumes map[string]any
+	execTime time.Time
+	mu       sync.Mutex
+}
+
+func (p *dagTestPlugin) PrepareRequestData(ctx context.Context, request *schedulingtypes.LLMRequest, pods []schedulingtypes.Pod) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.execTime = time.Now()
+	return p.mockPrepareRequestDataPlugin.PrepareRequestData(ctx, request, pods)
+}
+
+func (p *dagTestPlugin) Produces() map[string]any {
+	return p.produces
+}
+
+func (p *dagTestPlugin) Consumes() map[string]any {
+	return p.consumes
+}
+
+func TestExecutePluginsAsDAG(t *testing.T) {
+	pluginA := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "A", delay: 20 * time.Millisecond},
+		produces:                     map[string]any{"keyA": nil},
+	}
+	pluginB := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "B"},
+		consumes:                     map[string]any{"keyA": nil},
+		produces:                     map[string]any{"keyB": nil},
+	}
+	pluginC := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "C"},
+		consumes:                     map[string]any{"keyB": nil},
+	}
+	pluginD := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "D"},
+		consumes:                     map[string]any{"keyA": nil},
+	}
+	pluginE := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "E"},
+	}
+	pluginFail := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "Fail", returnErr: errors.New("plugin failed")},
+		produces:                     map[string]any{"keyFail": nil},
+	}
+	pluginDependsOnFail := &dagTestPlugin{
+		mockPrepareRequestDataPlugin: mockPrepareRequestDataPlugin{name: "DependsOnFail"},
+		consumes:                     map[string]any{"keyFail": nil},
+	}
+
+	testCases := []struct {
+		name      string
+		plugins   []PrepareDataPlugin
+		expectErr bool
+		checkFunc func(t *testing.T, plugins []PrepareDataPlugin)
+	}{
+		{
+			name:    "no plugins",
+			plugins: []PrepareDataPlugin{},
+		},
+		{
+			name:    "simple linear dependency (A -> B -> C)",
+			plugins: []PrepareDataPlugin{pluginA, pluginB, pluginC},
+			checkFunc: func(t *testing.T, plugins []PrepareDataPlugin) {
+				pA := plugins[0].(*dagTestPlugin)
+				pB := plugins[1].(*dagTestPlugin)
+				pC := plugins[2].(*dagTestPlugin)
+
+				assert.True(t, pA.executed, "Plugin A should have been executed")
+				assert.True(t, pB.executed, "Plugin B should have been executed")
+				assert.True(t, pC.executed, "Plugin C should have been executed")
+
+				assert.True(t, pB.execTime.After(pA.execTime), "Plugin B should execute after A")
+				assert.True(t, pC.execTime.After(pB.execTime), "Plugin C should execute after B")
+			},
+		},
+		{
+			name:    "DAG with multiple dependencies (A -> B, A -> D) and one independent (E)",
+			plugins: []PrepareDataPlugin{pluginA, pluginB, pluginD, pluginE},
+			checkFunc: func(t *testing.T, plugins []PrepareDataPlugin) {
+				pA := plugins[0].(*dagTestPlugin)
+				pB := plugins[1].(*dagTestPlugin)
+				pD := plugins[2].(*dagTestPlugin)
+				pE := plugins[3].(*dagTestPlugin)
+
+				assert.True(t, pA.executed, "Plugin A should have been executed")
+				assert.True(t, pB.executed, "Plugin B should have been executed")
+				assert.True(t, pD.executed, "Plugin D should have been executed")
+				assert.True(t, pE.executed, "Plugin E should have been executed")
+
+				assert.True(t, pB.execTime.After(pA.execTime), "Plugin B should execute after A")
+				assert.True(t, pD.execTime.After(pA.execTime), "Plugin D should execute after A")
+			},
+		},
+		{
+			name:      "dependency fails",
+			plugins:   []PrepareDataPlugin{pluginFail, pluginDependsOnFail},
+			expectErr: true,
+			checkFunc: func(t *testing.T, plugins []PrepareDataPlugin) {
+				pF := plugins[0].(*dagTestPlugin)
+				pDOF := plugins[1].(*dagTestPlugin)
+
+				assert.True(t, pF.executed, "Failing plugin should have been executed")
+				assert.False(t, pDOF.executed, "Plugin depending on fail should not be executed")
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Reset execution state for plugins
+			for _, p := range tc.plugins {
+				plugin := p.(*dagTestPlugin)
+				plugin.executed = false
+				plugin.execTime = time.Time{}
+			}
+
+			err := executePluginsAsDAG(tc.plugins, context.Background(), &schedulingtypes.LLMRequest{}, nil)
+
+			if tc.expectErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			if tc.checkFunc != nil {
+				tc.checkFunc(t, tc.plugins)
+			}
+		})
+	}
+}