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

Issue:
Prefix Cache Aware Proposal
We should have a proposal for how we will handle prefix-cache aware routing and have it accepted by the community in order to begin implementation

Hints from issue comments:
@liu-cong: I looked into this and here is a writeup on the problem and proposed solution. Please feel free to comment!

## Background

Prefix caching is a well-known technique in LLM inference to save duplicate tensor computation for prompts with the same prefixes, and is available in many model servers, as well as inference frameworks.

[vLLM](https://docs.vllm.ai/en/latest/features/automatic_prefix_caching.html) has the automatic prefix cache (APC) feature by caching in the accelerator HBM, and uses an LRU cache eviction strategy.

[vLLM production stack](https://github.com/vllm-project/production-stack/issues/59) is exploring a prefix aware router to exploit the APC feature of the vLLM. The WIP [PR](https://github.com/vllm-project/production-stack/issues/59#issuecomment-2677268482) implements two strategies: a HashTrie based matching and a SimHash based consistent hashing. The HashTrie solution is showing better cache hit rate.

[SGLang](https://github.com/sgl-project/sglang/blob/4d2a88bdffe91168dfc73ef7e3bc9100ba96686b/sgl-router/src/router.rs#L61) has a cache aware routing strategy which builds a radix tree based on request history.

[AIBrix](https://aibrix.readthedocs.io/latest/features/distributed-kv-cache.html) uses a distributed prefix cache pool and has a customized vLLM to support loading cache from the pool. At request routing, it has a [Prefix Router](https://github.com/vllm-project/aibrix/blob/6feec99d77c84e371da9c535054c2b8aa8912704/pkg/plugins/gateway/algorithms/prefix_cache.go#L64) that maximizes prefix cache hit on model server HBM. It currently implements a hash based (similar to vLLM) and radix tree based (similar to SGLang) matching strategy.

[KubeAI](https://www.kubeai.org/blog/2025/02/26/llm-load-balancing-at-scale-chwbl/) uses a Consistent Hashing with Bounded Loads (CHWBL)  algorithm which hashes request prefixes up to a configurable length (and therefore will lose some accuracy), and use an "overflow" strategy when the server is hot loaded.

## Goals
Implement a prefix aware routing algorithm on EPP to maximize the cache hit rate on the model servers.

**Non-goals**
* Change how model server manages prefix caches, e.g., add DRAM cache support or remote cache support.

## Design Options
### **1. Session affinity**

Session affinity is based on client attributes such as IP address. It works well for use cases such as multi-turn conversations, where requests from the same client tend to share the same prefixes. This, of course, highly depends on the nature of the use case.

Pros:
*   Easy to implement/understand

Cons:
*   Limited use case
*   Does not exploit prefix cache between different clients
*   Using client IP isn't always reliable, will likely need client to provide "session info" for good affinity

### **2. Prefix affinity consistent hashing**

This goes a step beyond the session affinity by using a prefix aware hash function to route requests with similar prefixes to the same or similar servers. A naive hash function can be just taking the hash of the first N characters/tokens of the request, and therefore all requests with the same first N characters/tokens will be routed to the same server. The [vLLM production stack](https://github.com/vllm-project/production-stack/issues/59) is exploring this strategy using simhash, and preliminary experiments showed mixed results. KubeAI uses a simple strategy to only hash request prefix up to a configurable `prefixCharLength`. Its effectiveness is likely highly dependent on the input length distribution.

Pros:

*   (Compared to session affinity) Is aware of prefix and not limited to per-client affinity
*   Small memory overhead (just need to store the ring of the servers)

Cons:
*   Highly depends on the effectiveness of the prefix aware hash function.
*   Consistent hashing can be challenging to reason about.
 
### **3. Approximate prefix cache on the router**
This builds on the intuition that if `requestA=prefix+XX` was routed to server 1, then routing `requstB=prefix+YY` to the same server will likely hit its prefix cache. Therefore the central router can build an approximate lookup cache of the prefix caches on all the backend servers, by mimicking a similar cache eviction strategy of the model server (e.g., LRU). 

Pros:
*   Easy to explain (compared to hashing) and likely more effective than hashing strategy.

Cons:
*   Relies on knowledge of the cache eviction strategy of the model server, and may need careful tuning for different environments (e.g., model server with different total kv cache space may have different characteristics of cache eviction).
*  Complexity in managing cache state (eviction, memory limit)
* An in memory cache is preferred for high performance. However, that means cache need to be rebuilt for restarts. Moreover, cache hit performance decreases with multiple active EPP replicas
 
### **4. Accurate prefix cache on the router**
If the router knows what prefixes are currently cached on each model server replica, it can make the optimal decision. A potential solution is to have the model server (or with a sidecar) report the kv cache indexes to the router.

Pros:
*   Best cache hit rate

Cons:
*   Requires adding a sidecar and its integration with the model server
*   May face scalability concerns with large number of model server replicas and large number of prefix caches


## How does prefix cache affinity routing work with LoRA affinity and load-aware routing

Prefix cache needs to be LoRA aware, as different adapters don’t share the same kv cache. Therefore, prefix cache affinity algo should be applied after LoRA affinity, to avoid conflicts. And by doing so, the requests will naturally be colocated by the same LoRA, and further by prefix matching.

Prefix affinity needs to be aware of the server load, otherwise we will create hot spots. To prevent this, a 2 step process can be used:

1. Only use prefix affinity if the expected gain is high enough. This can be done with a threshold, either on the prefix matching length, or a matching ratio of the entire request, or a combination of both. If the expected gain is below the threshold (e.g, only 10 tokens match), then we ignore prefix affinity and fall back to current load based algo.
2. Similar to the [queue depth threshold](https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/2a615e981228aa6ffc2a89219c986ac863dde776/pkg/epp/scheduling/scheduler.go#L40) for LoRA affinity, we define a similar one for prefix affinity. This prevents prefix affinity to overload a server indefinitely.


## Proposal 


**Preferred Approach**

Implement an approximate prefix cache lookup on the  EPP, similar to SGLang or AIBrix, potentially use a vLLM-alike strategy, i.e., use prefix block hash and an LRU cache eviction strategy.

A request is broken down into N chunks of the same number of characters (we don’t necessarily need to tokenize). For each chunk we will calculate a hash based on the **content of the chunk + hash of the prefix**: `hash(chunk i) = hash(chunk i content + hash(chunk i-1))`. 

When we route a request `r1` with `N` chunks to a server `s1`, we update the approximate cache lookup table like so:


```
hash(chunk 1): append s1
hash(chunk 2): append s1
…
hash(chunk N): append s1
```


This means all these N chunks are cached on server `s1`.

When the EPP receives a new request `r2`, we calculate its chunk hashes, and look up the table to find a server with longest prefix matching.

Each entry in the table needs a `lastUpdate` time to allow LRU cache eviction.

**Alternative Approach**

The key to the prefix aware consistent hashing is to find a great hash algorithm. [vLLM production stack](https://github.com/vllm-project/production-stack/issues/59) is exploring simhash, which is an algorithm used to detect similarities between sets, which is not quite exact as prefix matching. KubeAI uses a simple hash that only cares about request prefix to a certain limit.  I propose an improved algorithm which breaks the input into chunks and gives higher weight for chunks in the beginning. The result of this algorithm is that two inputs with more prefix matching should end up with closer hash results. However this needs careful tuning and benchmarking to validate.
@ahg-g: +1 to the preferred approach, this seems like a good first attempt that would provide tangible benefits. @liu-cong: 
## Implementation Plan


### PrefixMatcher

Conceptually the goal is to find the maximum prefix match of the input string. Instead of matching characters, we divide the string into chunks of size L to improve efficiency. The loss of accuracy is not important as long as we pick a reasonable L. Instead of storing the chunk characters, we calculate a hash value to reduce storage footprint using a high performance hash function such as xxhash.

For each chunk we will calculate a hash based on the** content of the chunk + hash of the prefix**: `hash(chunk i) = hash(chunk i content + hash(chunk i-1))`. This has the nice **property** that if `hash(request1, chunk i) == hash(request1, chunk i), `then we can claim they match prefixes at least until chunk i.

**Caveat: **Since different targetModels cannot share the same prefix cache, we will always **hash the targetModel in the first chunk**. 


#### Longest Prefix Matching

For an incoming request `r` with N chunks, we will implement a greedy search strategy by searching the cache look up table for the Nth chunk, and then (N-1)th, and so on. Note we can follow up with a binary search strategy to improve the search efficiency.


#### Update Cache

When we route a request `r1` with `N` chunks to a server `s1`, we update the approximate cache lookup table like so:


```
hash(r1, chunk 1): append s1
hash(r1, chunk 2): append s1
…
hash(r1, chunk N): append s1
```


This means all these N chunks are cached on server `s1`.

Concretely in go, the cache look up table is defined like so:


```
type PrefixCacheLookupTable struct {
        table map[BlockHash]CachedServers
        // Keep track of the total size of the table to avoid it exceeding a limit.
        size int
}
type BlockHash int64
// key: NamespacedName of the pod;
// value: The last time the corresponding chunk was sent to this server.
type CachedServers map[types.NamespacedName]time.Time
```



#### Cache Eviction

We spin up a goroutine which runs every `cacheEvictionInterval` to evict cache entries whose last accessed time exceeds `cacheExpiryDuration`.

Note: we can also leverage the cache eviction go routine to update the size of the cache.


#### Cache Size Cap

We need to ensure the cache size never exceeds a configurable `maxPrefixCacheSize`. If the size exceeds the limit, we simply stop adding new cache entries until old entries are evicted.


#### Implementation

We will create a prefixCacheMatcher in the `scheduling` package. 


```
type prefixMatcher struct {
   // cacheChunkSize is the size of each chunk in the cache. Requests with length shorter than the
   // chunk size will be ignored.
   cacheChunkSize int
   // goroutine interval to scan the cache and evict entries.
   cacheEvictionInterval time.Duration
   // cacheExpiration is the time after which the cache entry is considered expired.
   cacheExpiration time.Duration
   // If the cache size exceeds this limit, the cache will stop recording new entries.
   // Current (approximate) size of the cache.
   maxCacheSize int

   size         int
   table        *PrefixCacheLookupTable
}

func NewPrefixCacheMatcher() *prefixMatcher {
   return &prefixMatcher{}
}

// Given a request, and a list of candidate pods, look up the prefix cache table to find pods with the longest prefix match.
func (m *prefixMatcher) Match(request *LLMRequest, pods []backendmetrics.PodMetrics) ([]backendmetrics.PodMetrics, error) {
}

// If a request was routed to a pod, record it in the cache:
// Append the pod to the chunk hashes of the request. If the chunk was already cached in this pod, update it's last updated time.
func (m *prefixMatcher) Record(request *LLMRequest, pod backendmetrics.Pod) {
}

func (m *prefixMatcher) Evict(request *LLMRequest, pods []backendmetrics.PodMetrics) ([]backendmetrics.PodMetrics, error) {
   // Remove cache entries if last updated time is older than a threshold.
}
```



### Routing Algorithm

Prefix cache matching needs to be model specific, therefore we apply prefix cache after LoRA affinity. How does prefix caching logic interact with the `least queuing and least kv cache` logic? We define a `LoadAwarePrefixCacheFilter` that limits the prefix cache matching candidates to servers with load under a certain threshold.

In <code>[scheduler.go](http://scheduler.go)</code>, we define this:


```
   loadAwarePrefixCacheFilter = &filter{
       name: "load aware prefix cache filter",
       // Apply prefix caching filter to pods under load threshold if possible.
       // If all pods are above the threshold, then we apply prefix caching on all of them.
       filter: toFilterFunc(queueAndKVCacheThresholdPredicate(queueThreshold, kvCacheThreshold)), 
       nextOnSuccessOrFailure: &filter{
           name:   "pods with longest prefix cache match",
           filter: prefixCacheFilterFunc, // new filter func
       },
   }

```


This new filter will be applied after the LoRA affinity filter, and before any other filters.


### Configuration Knobs

In <code>[scheduler.go](http://scheduler.go)</code>, we will add


```
prefixCachingQueueThreshold // default to existing queueThresholdCritical (5)
prefixCachingKVCacheThreshold // default to existing kvCacheThreshold (0.8)
```


In <code>[prefixMatcher.go](http://prefixMatcher.go), </code>we will add


```
  // cacheChunkSize is the size of each chunk in the cache. Requests with length shorter than the
   // chunk size will be ignored.
   cacheChunkSize int
   // goroutine interval to scan the cache and evict entries.
   cacheEvictionInterval time.Duration
   // cacheExpiration is the time after which the cache entry is considered expired.
   cacheExpiration time.Duration
   // If the cache size (in MB) exceeds this limit, the cache will stop recording new entries.
   // Current (approximate) size of the cache.
   maxCacheSizeMB int
```

@smarterclayton: I would suggest from a scheduling perspective that we move to a standard filter and score approach as soon as possible - prefix matches are scoring, kvcache can be a filter or a score (hard or soft) or potentially both.  I would expect to let people have some flexibility to tweak score weighting in code before moving to make them tunable, for sure. @smarterclayton: I'll note another pro of the approximate approach is that we already have a need to track in flight requests and understand the delta of an given EPP's traffic from its peers in order to make better scheduling decisions (often called an assumption) - this is consistent with a general approach of having composable scheduling strategies.  We should be able to estimate a score that makes kv-cache awareness more accurate as well as estimating the cost of the request better.  @smarterclayton: I think this would be appropriate to move to a proposal PR so we can add comments inline and begin iterating.@liu-cong: > I think this would be appropriate to move to a proposal PR so we can add comments inline and begin iterating.

SG, preparing a PR for feedback@liu-cong: Drafted the proposal PR: https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/602

@liu-cong: /close 

https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/602@k8s-ci-robot: @liu-cong: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/498#issuecomment-2877142929):

>/close 
>
>https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/602


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/.gitignore b/.gitignore
index bd36bfb..4442b65 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,7 +7,6 @@
 bin/*
 Dockerfile.cross
 artifacts
-main
 
 # Test binary, built with `go test -c`
 *.test
diff --git a/Dockerfile b/Dockerfile
index 9cb62e2..acf05b7 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -16,11 +16,11 @@ COPY go.mod go.sum ./
 RUN go mod download
 
 # Sources
-COPY cmd ./cmd
-COPY pkg ./pkg
+COPY cmd/epp ./cmd
+COPY pkg/epp ./pkg/epp
 COPY internal ./internal
 COPY api ./api
-WORKDIR /src/cmd/epp
+WORKDIR /src/cmd
 RUN go build -ldflags="-X sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics.CommitSHA=${COMMIT_SHA}" -o /epp
 
 ## Multistage deploy
diff --git a/bbr.Dockerfile b/bbr.Dockerfile
index 03024e4..8f21eea 100644
--- a/bbr.Dockerfile
+++ b/bbr.Dockerfile
@@ -15,10 +15,10 @@ COPY go.mod go.sum ./
 RUN go mod download
 
 # Sources
-COPY cmd ./cmd
+COPY cmd/bbr ./cmd
 COPY pkg ./pkg
 COPY internal ./internal
-WORKDIR /src/cmd/bbr
+WORKDIR /src/cmd
 RUN go build -o /bbr
 
 ## Multistage deploy
diff --git a/cmd/bbr/health.go b/cmd/bbr/health.go
index 4ef8a8b..7d1b5fd 100644
--- a/cmd/bbr/health.go
+++ b/cmd/bbr/health.go
@@ -19,7 +19,6 @@ package main
 import (
 	"context"
 
-	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/go-logr/logr"
 	"google.golang.org/grpc/codes"
 	healthPb "google.golang.org/grpc/health/grpc_health_v1"
@@ -32,29 +31,10 @@ type healthServer struct {
 }
 
 func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckRequest) (*healthPb.HealthCheckResponse, error) {
-	if in.Service != extProcPb.ExternalProcessor_ServiceDesc.ServiceName {
-		s.logger.V(logutil.DEFAULT).Info("gRPC health check requested unknown service", "available-services", []string{extProcPb.ExternalProcessor_ServiceDesc.ServiceName}, "requested-service", in.Service)
-		return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVICE_UNKNOWN}, nil
-	}
-
-	s.logger.V(logutil.VERBOSE).Info("gRPC health check serving", "service", extProcPb.ExternalProcessor_ServiceDesc.ServiceName)
+	s.logger.V(logutil.VERBOSE).Info("gRPC health check serving", "service", in.Service)
 	return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVING}, nil
 }
 
-func (s *healthServer) List(ctx context.Context, _ *healthPb.HealthListRequest) (*healthPb.HealthListResponse, error) {
-	// currently only the ext_proc service is provided
-	serviceHealthResponse, err := s.Check(ctx, &healthPb.HealthCheckRequest{Service: extProcPb.ExternalProcessor_ServiceDesc.ServiceName})
-	if err != nil {
-		return nil, err
-	}
-
-	return &healthPb.HealthListResponse{
-		Statuses: map[string]*healthPb.HealthCheckResponse{
-			extProcPb.ExternalProcessor_ServiceDesc.ServiceName: serviceHealthResponse,
-		},
-	}, nil
-}
-
 func (s *healthServer) Watch(in *healthPb.HealthCheckRequest, srv healthPb.Health_WatchServer) error {
 	return status.Error(codes.Unimplemented, "Watch is not implemented")
 }
diff --git a/cmd/epp/health.go b/cmd/epp/health.go
index d5727d6..9369700 100644
--- a/cmd/epp/health.go
+++ b/cmd/epp/health.go
@@ -19,7 +19,6 @@ package main
 import (
 	"context"
 
-	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/go-logr/logr"
 	"google.golang.org/grpc/codes"
 	healthPb "google.golang.org/grpc/health/grpc_health_v1"
@@ -34,34 +33,14 @@ type healthServer struct {
 }
 
 func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckRequest) (*healthPb.HealthCheckResponse, error) {
-	if in.Service != extProcPb.ExternalProcessor_ServiceDesc.ServiceName {
-		s.logger.V(logutil.DEFAULT).Info("gRPC health check requested unknown service", "available-services", []string{extProcPb.ExternalProcessor_ServiceDesc.ServiceName}, "requested-service", in.Service)
-		return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVICE_UNKNOWN}, nil
-	}
-
 	if !s.datastore.PoolHasSynced() {
-		s.logger.V(logutil.DEFAULT).Info("gRPC health check not serving", "service", extProcPb.ExternalProcessor_ServiceDesc.ServiceName)
+		s.logger.V(logutil.DEFAULT).Info("gRPC health check not serving", "service", in.Service)
 		return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_NOT_SERVING}, nil
 	}
-
-	s.logger.V(logutil.TRACE).Info("gRPC health check serving", "service", extProcPb.ExternalProcessor_ServiceDesc.ServiceName)
+	s.logger.V(logutil.TRACE).Info("gRPC health check serving", "service", in.Service)
 	return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVING}, nil
 }
 
-func (s *healthServer) List(ctx context.Context, _ *healthPb.HealthListRequest) (*healthPb.HealthListResponse, error) {
-	// currently only the ext_proc service is provided
-	serviceHealthResponse, err := s.Check(ctx, &healthPb.HealthCheckRequest{Service: extProcPb.ExternalProcessor_ServiceDesc.ServiceName})
-	if err != nil {
-		return nil, err
-	}
-
-	return &healthPb.HealthListResponse{
-		Statuses: map[string]*healthPb.HealthCheckResponse{
-			extProcPb.ExternalProcessor_ServiceDesc.ServiceName: serviceHealthResponse,
-		},
-	}, nil
-}
-
 func (s *healthServer) Watch(in *healthPb.HealthCheckRequest, srv healthPb.Health_WatchServer) error {
 	return status.Error(codes.Unimplemented, "Watch is not implemented")
 }
diff --git a/cmd/epp/main.go b/cmd/epp/main.go
index 9fd401d..bda7cc2 100644
--- a/cmd/epp/main.go
+++ b/cmd/epp/main.go
@@ -34,6 +34,7 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/component-base/metrics/legacyregistry"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
@@ -43,7 +44,13 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics/collectors"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/filter"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/picker"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/prefix"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/scorer"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
+	envutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/env"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
@@ -107,8 +114,22 @@ var (
 		"Prometheus metric for the LoRA info metrics (must be in vLLM label format).")
 
 	setupLog = ctrl.Log.WithName("setup")
+
+	// Environment variables
+	schedulerV2           = envutil.GetEnvString("EXPERIMENTAL_USE_SCHEDULER_V2", "false", setupLog)
+	prefixCacheScheduling = envutil.GetEnvString("ENABLE_PREFIX_CACHE_SCHEDULING", "false", setupLog)
 )
 
+func loadPrefixCacheConfig() prefix.Config {
+	baseLogger := log.Log.WithName("env-config")
+
+	return prefix.Config{
+		HashBlockSize:          envutil.GetEnvInt("PREFIX_CACHE_HASH_BLOCK_SIZE", prefix.DefaultHashBlockSize, baseLogger),
+		MaxPrefixBlocksToMatch: envutil.GetEnvInt("PREFIX_CACHE_MAX_PREFIX_BLOCKS", prefix.DefaultMaxPrefixBlocks, baseLogger),
+		LRUIndexerCapacity:     envutil.GetEnvInt("PREFIX_CACHE_LRU_CAPACITY", prefix.DefaultLRUIndexerCapacity, baseLogger),
+	}
+}
+
 func main() {
 	if err := run(); err != nil {
 		os.Exit(1)
@@ -172,6 +193,28 @@ func run() error {
 	datastore := datastore.NewDatastore(ctx, pmf)
 
 	scheduler := scheduling.NewScheduler(datastore)
+	if schedulerV2 == "true" {
+		queueScorerWeight := envutil.GetEnvInt("QUEUE_SCORE_WEIGHT", scorer.DefaultQueueScorerWeight, setupLog)
+		kvCacheScorerWeight := envutil.GetEnvInt("KV_CACHE_SCORE_WEIGHT", scorer.DefaultKVCacheScorerWeight, setupLog)
+		scorers := map[plugins.Scorer]int{
+			&scorer.QueueScorer{}:   queueScorerWeight,
+			&scorer.KVCacheScorer{}: kvCacheScorerWeight,
+		}
+		schedConfigOpts := []scheduling.ConfigOption{}
+		if prefixCacheScheduling == "true" {
+			prefixScorerWeight := envutil.GetEnvInt("PREFIX_CACHE_SCORE_WEIGHT", prefix.DefaultScorerWeight, setupLog)
+			schedConfigOpts = append(schedConfigOpts, scheduling.AddPrefixPlugin(loadPrefixCacheConfig(), prefixScorerWeight))
+		}
+		schedulerConfig := scheduling.NewSchedulerConfig(
+			[]plugins.PreSchedule{},
+			[]plugins.Filter{filter.NewSheddableCapacityFilter()},
+			scorers,
+			picker.NewMaxScorePicker(),
+			[]plugins.PostSchedule{},
+			[]plugins.PostResponse{},
+			schedConfigOpts...)
+		scheduler = scheduling.NewSchedulerWithConfig(datastore, schedulerConfig)
+	}
 	serverRunner := &runserver.ExtProcServerRunner{
 		GrpcPort:                                 *grpcPort,
 		DestinationEndpointHintMetadataNamespace: *destinationEndpointHintMetadataNamespace,
diff --git a/config/charts/inferencepool/templates/epp-deployment.yaml b/config/charts/inferencepool/templates/epp-deployment.yaml
index c391d06..fc49021 100644
--- a/config/charts/inferencepool/templates/epp-deployment.yaml
+++ b/config/charts/inferencepool/templates/epp-deployment.yaml
@@ -53,12 +53,12 @@ spec:
         livenessProbe:
           grpc:
             port: 9003
-            service: envoy.service.ext_proc.v3.ExternalProcessor
+            service: inference-extension
           initialDelaySeconds: 5
           periodSeconds: 10
         readinessProbe:
           grpc:
             port: 9003
-            service: envoy.service.ext_proc.v3.ExternalProcessor
+            service: inference-extension
           initialDelaySeconds: 5
           periodSeconds: 10
diff --git a/config/manifests/benchmark/benchmark.yaml b/config/manifests/benchmark/benchmark.yaml
index c784730..abf9ae5 100644
--- a/config/manifests/benchmark/benchmark.yaml
+++ b/config/manifests/benchmark/benchmark.yaml
@@ -37,7 +37,7 @@ spec:
         - name: BACKEND
           value: vllm
         - name: PORT
-          value: "8081"
+          value: "80"
         - name: INPUT_LENGTH
           value: "1024"
         - name: OUTPUT_LENGTH
diff --git a/config/manifests/inferencepool-resources.yaml b/config/manifests/inferencepool-resources.yaml
index e1b359e..37099c4 100644
--- a/config/manifests/inferencepool-resources.yaml
+++ b/config/manifests/inferencepool-resources.yaml
@@ -71,13 +71,13 @@ spec:
         livenessProbe:
           grpc:
             port: 9003
-            service: envoy.service.ext_proc.v3.ExternalProcessor
+            service: inference-extension
           initialDelaySeconds: 5
           periodSeconds: 10
         readinessProbe:
           grpc:
             port: 9003
-            service: envoy.service.ext_proc.v3.ExternalProcessor
+            service: inference-extension
           initialDelaySeconds: 5
           periodSeconds: 10
 ---
diff --git a/conformance/conformance.go b/conformance/conformance.go
index 20d80fd..1e847fd 100644
--- a/conformance/conformance.go
+++ b/conformance/conformance.go
@@ -19,30 +19,38 @@ limitations under the License.
 package conformance
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"io/fs"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
+	clientsetscheme "k8s.io/client-go/kubernetes/scheme"
 
 	// Import runtime package for scheme creation
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	k8sconfig "sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/yaml"
 
 	// Import necessary types and utilities from the core Gateway API conformance suite.
-	// Assumes sigs.k8s.io/gateway-api is a dependency in the go.mod.
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"            // Import core Gateway API types
 	confapis "sigs.k8s.io/gateway-api/conformance/apis/v1" // Report struct definition
 	confconfig "sigs.k8s.io/gateway-api/conformance/utils/config"
 	confflags "sigs.k8s.io/gateway-api/conformance/utils/flags"
+	apikubernetes "sigs.k8s.io/gateway-api/conformance/utils/kubernetes"
 	confsuite "sigs.k8s.io/gateway-api/conformance/utils/suite"
-	"sigs.k8s.io/gateway-api/pkg/features" // Using core features definitions if applicable
+	"sigs.k8s.io/gateway-api/pkg/features"
 
 	// Import the test definitions package to access the ConformanceTests slice
 	"sigs.k8s.io/gateway-api-inference-extension/conformance/tests"
@@ -58,48 +66,59 @@ import (
 	inferencev1alpha2 "sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
 )
 
+// Constants for the shared Gateway
+const (
+	SharedGatewayName      = "conformance-gateway"       // Name of the Gateway in manifests.yaml
+	SharedGatewayNamespace = "gateway-conformance-infra" // Namespace of the Gateway
+)
+
 // GatewayLayerProfileName defines the name for the conformance profile that tests
 // the Gateway API layer aspects of the Inference Extension (e.g., InferencePool, InferenceModel CRDs).
 // Future profiles will cover EPP and ModelServer layers.
 const GatewayLayerProfileName confsuite.ConformanceProfileName = "Gateway"
 
-var InferenceCoreFeatures = sets.New[features.FeatureName]() // Placeholder - Populate with actual features specific to this profile or manage features per profile
+// InferenceCoreFeatures defines the core features that implementations
+// of the "Gateway" profile for the Inference Extension MUST support.
+var InferenceCoreFeatures = sets.New(
+	features.SupportGateway, // This is needed to ensure manifest gets applied during setup.
+)
 
-// GatewayLayerProfile defines the conformance profile for the Gateway API layer
-// of the Inference Extension.
-// In future iterations, we will add constants and ConformanceProfile structs for
-// EPPProfileName ("EPP") and ModelServerProfileName ("ModelServer")
-// to cover their respective conformance layers.
 var GatewayLayerProfile = confsuite.ConformanceProfile{
 	Name:         GatewayLayerProfileName,
 	CoreFeatures: InferenceCoreFeatures,
 }
 
+// logDebugf conditionally logs a debug message if debug mode is enabled.
+func logDebugf(t *testing.T, debug bool, format string, args ...any) {
+	if debug {
+		t.Helper()
+		t.Logf(format, args...)
+	}
+}
+
 // DefaultOptions parses command line flags and sets up the suite options.
 // Adapted from the core Gateway API conformance suite.
 func DefaultOptions(t *testing.T) confsuite.ConformanceOptions {
 	t.Helper()
 
-	cfg, err := config.GetConfig()
+	cfg, err := k8sconfig.GetConfig()
 	require.NoError(t, err, "error loading Kubernetes config")
 
-	// Initialize client options. The scheme must include Gateway API types
-	// and the Inference Extension types.
-	clientOptions := client.Options{}
-	scheme := clientOptions.Scheme
-	if scheme == nil {
-		// If default options don't provide a scheme, create one using runtime.NewScheme().
-		scheme = runtime.NewScheme()
-		clientOptions.Scheme = scheme
-	}
+	scheme := runtime.NewScheme()
+
+	t.Log("Registering API types with scheme...")
+	// Register core K8s types (like v1.Secret for certs) to scheme, needed by client to create/manage these resources.
+	require.NoError(t, clientsetscheme.AddToScheme(scheme), "failed to add core Kubernetes types to scheme")
+	// Add Gateway API types
+	require.NoError(t, gatewayv1.Install(scheme), "failed to install gatewayv1 types into scheme")
+	// Add APIExtensions types (for CRDs)
+	require.NoError(t, apiextensionsv1.AddToScheme(scheme), "failed to add apiextensionsv1 types to scheme")
 
-	// Register necessary API Types
-	require.NoError(t, gatewayv1.Install(scheme)) // Add core Gateway API types
-	// Add the Inference Extension API types to the scheme using the correct import alias
-	require.NoError(t, inferencev1alpha2.Install(scheme))
-	require.NoError(t, apiextensionsv1.AddToScheme(scheme)) // Needed for CRD checks
+	// Register Inference Extension API types
+	t.Logf("Attempting to install inferencev1alpha2 types into scheme from package: %s", inferencev1alpha2.GroupName)
+	require.NoError(t, inferencev1alpha2.Install(scheme), "failed to install inferencev1alpha2 types into scheme")
 
-	// Create the Kubernetes clients
+	clientOptions := client.Options{Scheme: scheme}
 	c, err := client.New(cfg, clientOptions)
 	require.NoError(t, err, "error initializing Kubernetes client")
 	cs, err := clientset.NewForConfig(cfg)
@@ -124,15 +143,18 @@ func DefaultOptions(t *testing.T) confsuite.ConformanceOptions {
 	inferenceExtensionVersion := "v0.3.0"
 	_ = inferenceExtensionVersion // Avoid unused variable error until implemented
 
-	// Create ConformanceOptions
+	baseManifestsValue := "resources/manifests/manifests.yaml"
+
 	opts := confsuite.ConformanceOptions{
 		Client:               c,
+		ClientOptions:        clientOptions,
 		Clientset:            cs,
 		RestConfig:           cfg,
 		GatewayClassName:     *confflags.GatewayClassName,
+		BaseManifests:        baseManifestsValue,
 		Debug:                *confflags.ShowDebug,
 		CleanupBaseResources: *confflags.CleanupBaseResources,
-		SupportedFeatures:    sets.New[features.FeatureName](), // Initialize empty, will be populated below
+		SupportedFeatures:    sets.New[features.FeatureName](),
 		TimeoutConfig:        confconfig.DefaultTimeoutConfig(),
 		SkipTests:            skipTests,
 		ExemptFeatures:       exemptFeatures,
@@ -140,7 +162,7 @@ func DefaultOptions(t *testing.T) confsuite.ConformanceOptions {
 		Mode:                 *confflags.Mode,
 		Implementation:       implementation,
 		ConformanceProfiles:  conformanceProfiles,
-		ManifestFS:           []fs.FS{&Manifests}, // Assumes embed.go defines `Manifests`
+		ManifestFS:           []fs.FS{&Manifests},
 		ReportOutputPath:     *confflags.ReportOutput,
 		SkipProvisionalTests: *confflags.SkipProvisionalTests,
 		// TODO: Add the inference extension specific fields to ConformanceOptions struct if needed,
@@ -152,16 +174,20 @@ func DefaultOptions(t *testing.T) confsuite.ConformanceOptions {
 	// Populate SupportedFeatures based on the GatewayLayerProfile.
 	// Since all features are mandatory for this profile, add all defined core features.
 	if opts.ConformanceProfiles.Has(GatewayLayerProfileName) {
-		for feature := range GatewayLayerProfile.CoreFeatures {
-			opts.SupportedFeatures.Insert(feature)
+		logDebugf(t, opts.Debug, "Populating SupportedFeatures with GatewayLayerProfile.CoreFeatures: %v", GatewayLayerProfile.CoreFeatures.UnsortedList())
+		if GatewayLayerProfile.CoreFeatures.Len() > 0 {
+			opts.SupportedFeatures = opts.SupportedFeatures.Insert(GatewayLayerProfile.CoreFeatures.UnsortedList()...)
 		}
 	}
 
 	// Remove any features explicitly exempted via flags.
-	for feature := range opts.ExemptFeatures {
-		opts.SupportedFeatures.Delete(feature)
+	if opts.ExemptFeatures.Len() > 0 {
+		logDebugf(t, opts.Debug, "Removing ExemptFeatures from SupportedFeatures: %v", opts.ExemptFeatures.UnsortedList())
+		opts.SupportedFeatures = opts.SupportedFeatures.Delete(opts.ExemptFeatures.UnsortedList()...)
 	}
 
+	logDebugf(t, opts.Debug, "Final opts.SupportedFeatures: %v", opts.SupportedFeatures.UnsortedList())
+
 	return opts
 }
 
@@ -172,7 +198,9 @@ func RunConformance(t *testing.T) {
 
 // RunConformanceWithOptions runs the Inference Extension conformance tests with specific options.
 func RunConformanceWithOptions(t *testing.T, opts confsuite.ConformanceOptions) {
+	t.Helper()
 	t.Logf("Running Inference Extension conformance tests with GatewayClass %s", opts.GatewayClassName)
+	logDebugf(t, opts.Debug, "RunConformanceWithOptions: BaseManifests path being used by opts: %q", opts.BaseManifests)
 
 	// Register the GatewayLayerProfile with the suite runner.
 	// In the future, other profiles (EPP, ModelServer) will also be registered here,
@@ -183,13 +211,13 @@ func RunConformanceWithOptions(t *testing.T, opts confsuite.ConformanceOptions)
 	cSuite, err := confsuite.NewConformanceTestSuite(opts)
 	require.NoError(t, err, "error initializing conformance suite")
 
-	t.Log("Setting up Inference Extension conformance tests")
-	// Setup requires the list of tests, which is populated by the init() functions
-	// triggered by the blank imports at the top of this file.
 	cSuite.Setup(t, tests.ConformanceTests)
 
-	t.Log("Running Inference Extension conformance tests")
-	// Run the tests.
+	sharedGwNN := types.NamespacedName{Name: SharedGatewayName, Namespace: SharedGatewayNamespace}
+
+	// Validate Gateway setup.
+	ensureGatewayAvailableAndReady(t, cSuite.Client, opts, sharedGwNN)
+	t.Log("Running Inference Extension conformance tests against all registered tests")
 	err = cSuite.Run(t, tests.ConformanceTests)
 	require.NoError(t, err, "error running conformance tests")
 
@@ -209,6 +237,67 @@ func RunConformanceWithOptions(t *testing.T, opts confsuite.ConformanceOptions)
 	}
 }
 
+// ensureGatewayAvailableAndReady polls for the specified Gateway to exist and become ready
+// with an address and programmed condition.
+func ensureGatewayAvailableAndReady(t *testing.T, k8sClient client.Client, opts confsuite.ConformanceOptions, gatewayNN types.NamespacedName) {
+	t.Helper()
+
+	t.Logf("Attempting to fetch Gateway %s/%s.", gatewayNN.Namespace, gatewayNN.Name)
+	gw := &gatewayv1.Gateway{} // This gw instance will be populated by the poll function
+
+	// Define polling interval
+	// TODO: Make this configurable using a local TimeoutConfig (from ConformanceOptions perhaps)
+	pollingInterval := 5 * time.Second
+	// Use the GatewayMustHaveAddress timeout from the suite's TimeoutConfig for the Gateway object to appear
+	waitForGatewayCreationTimeout := opts.TimeoutConfig.GatewayMustHaveAddress
+
+	logDebugf(t, opts.Debug, "Waiting up to %v for Gateway object %s/%s to appear after manifest application...", waitForGatewayCreationTimeout, gatewayNN.Namespace, gatewayNN.Name)
+
+	ctx := context.TODO()
+	pollErr := wait.PollUntilContextTimeout(ctx, pollingInterval, waitForGatewayCreationTimeout, true, func(pollCtx context.Context) (bool, error) {
+		fetchErr := k8sClient.Get(pollCtx, gatewayNN, gw)
+		if fetchErr == nil {
+			t.Logf("Successfully fetched Gateway %s/%s. Spec.GatewayClassName: %s",
+				gw.Namespace, gw.Name, gw.Spec.GatewayClassName)
+			return true, nil
+		}
+		if apierrors.IsNotFound(fetchErr) {
+			logDebugf(t, opts.Debug, "Gateway %s/%s not found, still waiting...", gatewayNN.Namespace, gatewayNN.Name)
+			return false, nil // Not found, continue polling
+		}
+		// For any other error, stop polling and return this error
+		t.Logf("Error fetching Gateway %s/%s: %v. Halting polling for this attempt.", gatewayNN.Namespace, gatewayNN.Name, fetchErr)
+		return false, fetchErr
+	})
+
+	// Check if polling timed out or an error occurred during polling
+	if pollErr != nil {
+		var failureMessage string
+		if errors.Is(pollErr, context.DeadlineExceeded) {
+			failureMessage = fmt.Sprintf("Timed out after %v waiting for Gateway object %s/%s to appear in the API server.",
+				waitForGatewayCreationTimeout, gatewayNN.Namespace, gatewayNN.Name)
+		} else {
+			failureMessage = fmt.Sprintf("Error while waiting for Gateway object %s/%s to appear: %v.",
+				gatewayNN.Namespace, gatewayNN.Name, pollErr)
+		}
+		finalMessage := failureMessage + " The Gateway object should have been created by the base manifest application."
+		require.FailNow(t, finalMessage) // Use FailNow to stop if the Gateway isn't found.
+	}
+
+	logDebugf(t, opts.Debug, "Waiting for shared Gateway %s/%s to be ready", gatewayNN.Namespace, gatewayNN.Name)
+	apikubernetes.GatewayMustHaveCondition(t, k8sClient, opts.TimeoutConfig, gatewayNN, metav1.Condition{
+		Type:   string(gatewayv1.GatewayConditionAccepted),
+		Status: metav1.ConditionTrue,
+	})
+	apikubernetes.GatewayMustHaveCondition(t, k8sClient, opts.TimeoutConfig, gatewayNN, metav1.Condition{
+		Type:   string(gatewayv1.GatewayConditionProgrammed),
+		Status: metav1.ConditionTrue,
+	})
+	_, err := apikubernetes.WaitForGatewayAddress(t, k8sClient, opts.TimeoutConfig, apikubernetes.NewGatewayRef(gatewayNN))
+	require.NoErrorf(t, err, "shared gateway %s/%s did not get an address", gatewayNN.Namespace, gatewayNN.Name)
+	t.Logf("Shared Gateway %s/%s is ready.", gatewayNN.Namespace, gatewayNN.Name)
+}
+
 // writeReport writes the generated conformance report to the specified output file or logs it.
 // Adapted from the core Gateway API suite.
 func writeReport(logf func(string, ...any), report confapis.ConformanceReport, output string) error {
diff --git a/conformance/embed.go b/conformance/embed.go
index f7fa64c..c9175db 100644
--- a/conformance/embed.go
+++ b/conformance/embed.go
@@ -21,5 +21,5 @@ import "embed"
 // Manifests embeds the contents of the conformance/resources directory making
 // the YAML files within them available to the test suite at runtime.
 //
-//go:embed resources/* tests/*
+//go:embed resources tests/*
 var Manifests embed.FS
diff --git a/conformance/resources/manifests/manifests.yaml b/conformance/resources/manifests/manifests.yaml
index 7b43b78..190a884 100644
--- a/conformance/resources/manifests/manifests.yaml
+++ b/conformance/resources/manifests/manifests.yaml
@@ -22,15 +22,23 @@ metadata:
   labels:
     gateway-conformance: backend
 
+---
+# Namespace for simple web server backends. This is expected by 
+# the upstream conformance suite's Setup method.
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: gateway-conformance-web-backend
+  labels:
+    gateway-conformance: web-backend
+
 ---
 # A basic Gateway resource that allows HTTPRoutes from the same namespace.
 # Tests can use this as a parent reference for routes that target InferencePools.
-# Using a simple echo server instead of an actual model server to simplify the test  
-# execution, this design may need to be revised based on the test case needs. 
-apiVersion: gateway.networking.k8s.io/v1 # Using v1 as per latest Gateway API standard
+apiVersion: gateway.networking.k8s.io/v1
 kind: Gateway
 metadata:
-  name: same-namespace
+  name: conformance-gateway
   namespace: gateway-conformance-infra
 spec:
   # The conformance suite runner will replace this placeholder
@@ -42,7 +50,7 @@ spec:
     protocol: HTTP
     allowedRoutes:
       namespaces:
-        from: Same # Restrict to same namespace initially for simplicity
+        from: All 
       kinds:
       # Allows HTTPRoutes to attach, which can then reference InferencePools.
       - group: gateway.networking.k8s.io
diff --git a/conformance/utils/kubernetes/helpers.go b/conformance/utils/kubernetes/helpers.go
index 3d51786..af7d5a2 100644
--- a/conformance/utils/kubernetes/helpers.go
+++ b/conformance/utils/kubernetes/helpers.go
@@ -19,31 +19,134 @@ limitations under the License.
 package kubernetes
 
 import (
+	"context"
+	"fmt"
+	"reflect"
 	"testing"
+	"time"
 
+	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	// Import the Inference Extension API types
+	inferenceapi "sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2" // Adjust if your API version is different
+
 	// Import necessary utilities from the core Gateway API conformance suite
 	"sigs.k8s.io/gateway-api/conformance/utils/config"
 )
 
+// checkCondition is a helper function similar to findConditionInList or CheckCondition
+// from the Gateway API conformance utilities.
+// It checks if the expectedCondition is present in the conditions list.
+// If expectedCondition.Reason is an empty string, it matches any reason.
+func checkCondition(t *testing.T, conditions []metav1.Condition, expectedCondition metav1.Condition) bool {
+	t.Helper()
+	for _, cond := range conditions {
+		if cond.Type == expectedCondition.Type {
+			if cond.Status == expectedCondition.Status {
+				if expectedCondition.Reason == "" || cond.Reason == expectedCondition.Reason {
+					return true
+				}
+				t.Logf("Condition %s found with Status %s, but Reason %s did not match expected %s",
+					expectedCondition.Type, cond.Status, cond.Reason, expectedCondition.Reason)
+			} else {
+				t.Logf("Condition %s found, but Status %s did not match expected %s",
+					expectedCondition.Type, cond.Status, expectedCondition.Status)
+			}
+		}
+	}
+	t.Logf("Condition %s with Status %s (and Reason %s if specified) not found in conditions list: %+v",
+		expectedCondition.Type, expectedCondition.Status, expectedCondition.Reason, conditions)
+	return false
+}
+
 // InferencePoolMustHaveCondition waits for the specified InferencePool resource
-// to exist and report the expected status condition.
-// This is a placeholder and needs full implementation.
-//
-// TODO: Implement the actual logic for this helper function.
-// It should fetch the InferencePool using the provided client and check its
-// Status.Conditions field, polling until the condition is met or a timeout occurs.
-// like HTTPRouteMustHaveCondition.
+// to exist and report the expected status condition within one of its parent statuses.
+// It polls the InferencePool's status until the condition is met or the timeout occurs.
 func InferencePoolMustHaveCondition(t *testing.T, c client.Client, timeoutConfig config.TimeoutConfig, poolNN types.NamespacedName, expectedCondition metav1.Condition) {
 	t.Helper() // Marks this function as a test helper
 
-	// Placeholder implementation: Log and skip the check.
-	t.Logf("Verification for InferencePool condition (%s=%s) on %s - Placeholder: Skipping check.",
-		expectedCondition.Type, expectedCondition.Status, poolNN.String())
+	var lastObservedPool *inferenceapi.InferencePool
+	var lastError error
+	var conditionFound bool
+	var interval time.Duration = 5 * time.Second // pull interval for status checks.
+
+	// TODO: Make retry interval configurable.
+	waitErr := wait.PollUntilContextTimeout(context.Background(), interval, timeoutConfig.DefaultTestTimeout, true, func(ctx context.Context) (bool, error) {
+		pool := &inferenceapi.InferencePool{} // This is the type instance used for Get
+		err := c.Get(ctx, poolNN, pool)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				t.Logf("InferencePool %s not found yet. Retrying.", poolNN.String())
+				lastError = err
+				return false, nil
+			}
+			t.Logf("Error fetching InferencePool %s (type: %s): %v. Retrying.", poolNN.String(), reflect.TypeOf(pool).String(), err)
+			lastError = err
+			return false, nil
+		}
+		lastObservedPool = pool
+		lastError = nil
+		conditionFound = false
+
+		if len(pool.Status.Parents) == 0 {
+			t.Logf("InferencePool %s has no parent statuses reported yet.", poolNN.String())
+			return false, nil
+		}
+
+		for _, parentStatus := range pool.Status.Parents {
+			if checkCondition(t, parentStatus.Conditions, expectedCondition) {
+				conditionFound = true
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+
+	if waitErr != nil || !conditionFound {
+		debugMsg := ""
+		if waitErr != nil {
+			debugMsg += fmt.Sprintf(" Polling error: %v.", waitErr)
+		}
+		if lastError != nil {
+			debugMsg += fmt.Sprintf(" Last error during fetching: %v.", lastError)
+		}
+
+		if lastObservedPool != nil {
+			debugMsg += "\nLast observed InferencePool status:"
+			if len(lastObservedPool.Status.Parents) == 0 {
+				debugMsg += " (No parent statuses reported)"
+			}
+			for i, parentStatus := range lastObservedPool.Status.Parents {
+				debugMsg += fmt.Sprintf("\n  Parent %d (Gateway: %s/%s):", i, parentStatus.GatewayRef.Namespace, parentStatus.GatewayRef.Name)
+				if len(parentStatus.Conditions) == 0 {
+					debugMsg += " (No conditions reported for this parent)"
+				}
+				for _, cond := range parentStatus.Conditions {
+					debugMsg += fmt.Sprintf("\n    - Type: %s, Status: %s, Reason: %s, Message: %s", cond.Type, cond.Status, cond.Reason, cond.Message)
+				}
+			}
+		} else if lastError == nil || !apierrors.IsNotFound(lastError) {
+			debugMsg += "\nInferencePool was not found or not observed successfully during polling."
+		}
+
+		finalMsg := fmt.Sprintf("timed out or condition not met for InferencePool %s to have condition Type=%s, Status=%s",
+			poolNN.String(), expectedCondition.Type, expectedCondition.Status)
+		if expectedCondition.Reason != "" {
+			finalMsg += fmt.Sprintf(", Reason='%s'", expectedCondition.Reason)
+		}
+		finalMsg += "." + debugMsg
+		require.FailNow(t, finalMsg)
+	}
 
-	// Skip the test using this helper until it's fully implemented.
-	t.Skip("InferencePoolMustHaveCondition helper not yet implemented")
+	logMsg := fmt.Sprintf("InferencePool %s successfully has condition Type=%s, Status=%s",
+		poolNN.String(), expectedCondition.Type, expectedCondition.Status)
+	if expectedCondition.Reason != "" {
+		logMsg += fmt.Sprintf(", Reason='%s'", expectedCondition.Reason)
+	}
+	t.Log(logMsg)
 }
diff --git a/docs/proposals/0602-prefix-cache-aware-routing-proposal/README.md b/docs/proposals/0602-prefix-cache-aware-routing-proposal/README.md
new file mode 100644
index 0000000..468e3be
--- /dev/null
+++ b/docs/proposals/0602-prefix-cache-aware-routing-proposal/README.md
@@ -0,0 +1,124 @@
+# Prefix Cache Aware Request Scheduling
+
+## Overview
+
+Prefix caching is a well-known technique in LLM inference to save duplicate tensor computation for prompts with the same prefix tokens, and is available in many model servers or model as a service providers. Leveraging prefix caching can significantly boost system performance, especially the time to first token (TTFT). Given that EPP has a global view of requests and model servers in the `InferencePool`, it can schedule requests intelligently to maximize the global prefix cache hit rate.
+
+### Goals
+
+Implement a prefix aware scheduling algorithm on EPP to maximize the cache hit rate on the model servers.
+
+### Non-goals
+
+* Change how model server manages prefix caches, or add any prefix cache APIs.
+* Coordinate cache beyond accelerator HBM cache, such as remote caches.
+
+## Terminology
+
+In the gateway-api-inference-extension project, we use the term "request scheduling" to mean the process of estimating the cost of a request and placing it to the best backend server. This is different from "model routing" which oftentimes means picking the right model server endpoint based on cost, availability, etc. However, we acknowledge that various other projects uses the term "routing" or "router" to mean what we call "request scheduling". In this doc, we use "scheduling" when referring to the inference extension, and "routing" or "router" when referring to other projects, respecting the terminology of those projects.
+
+## Existing Solutions
+
+[vLLM](https://docs.vllm.ai/en/latest/features/automatic_prefix_caching.html) has the automatic prefix cache (APC) feature by caching in the accelerator HBM, and uses an LRU cache eviction strategy.
+
+[vLLM production stack](https://github.com/vllm-project/production-stack/issues/59) is exploring a prefix aware router to exploit the APC feature of the vLLM. The WIP [PR](https://github.com/vllm-project/production-stack/issues/59#issuecomment-2677268482) implements two strategies: a HashTrie based matching and a SimHash based consistent hashing. The HashTrie solution is showing better cache hit rate.
+
+[SGLang](https://github.com/sgl-project/sglang/blob/4d2a88bdffe91168dfc73ef7e3bc9100ba96686b/sgl-router/src/router.rs#L61) has a cache aware routing strategy which builds a radix tree based on request history.
+
+[AIBrix](https://aibrix.readthedocs.io/latest/features/distributed-kv-cache.html) uses a distributed prefix cache pool and has a customized vLLM to support loading cache from the pool. At request routing, it has a [Prefix Router](https://github.com/vllm-project/aibrix/blob/6feec99d77c84e371da9c535054c2b8aa8912704/pkg/plugins/gateway/algorithms/prefix_cache.go#L64) that maximizes prefix cache hit on model server HBM. It currently implements a hash based (similar to vLLM) and radix tree based (similar to SGLang) matching strategy.
+
+[KubeAI](https://www.kubeai.org/blog/2025/02/26/llm-load-balancing-at-scale-chwbl/) uses a Consistent Hashing with Bounded Loads (CHWBL)  algorithm which hashes request prefixes up to a configurable length (and therefore will lose some accuracy), and use an "overflow" strategy when the server is hot loaded.
+
+## Design Options
+
+### Session affinity
+
+Session affinity is based on client attributes such as IP address. It works well for use cases such as multi-turn conversations, where requests from the same client tend to share the same prefixes. This, of course, highly depends on the nature of the use case.
+
+Pros:
+
+* Easy to implement/understand
+
+Cons:
+
+* Limited use case
+* Does not exploit prefix cache between different clients
+* Using client IP isn't always reliable, will likely need client to provide "session info" for good affinity
+
+### Prefix affinity consistent hashing
+
+This goes a step beyond the session affinity by using a prefix aware hash function to schedule requests with similar prefixes to the same or similar servers. A naive hash function can be just taking the hash of the first N characters/tokens of the request, and therefore all requests with the same first N characters/tokens will be scheduled to the same server. The [vLLM production stack](https://github.com/vllm-project/production-stack/issues/59) is exploring this strategy using simhash, and preliminary experiments showed mixed results. KubeAI uses a simple strategy to only hash request prefix up to a configurable `prefixCharLength`. Its effectiveness is likely highly dependent on the input length distribution.
+
+Pros:
+
+* (Compared to session affinity) Is aware of prefix and not limited to per-client affinity
+* Small memory overhead (just need to store the ring of the servers)
+
+Cons:
+
+* Highly depends on the effectiveness of the prefix aware hash function.
+* Consistent hashing can be challenging to reason about.
+ 
+### Report prefix cache indexes on the EPP
+
+If the EPP knows what prefixes are currently cached on each model server replica, it can make the optimal decision. A potential solution is to have the model server (or with a sidecar) report the kv cache indexes to the EPP.
+
+Pros:
+
+* Best cache hit rate in theory
+
+Cons:
+
+* Requires API changes on the model servers to report the cache indexes.
+* Reporting the cache indexes in real time requires non-trivial network bandwidth.
+
+### Approximate prefix index on the EPP
+
+This builds on the intuition that if `requestA=prefix+XX` was scheduled to server 1, then scheduling `requestB=prefix+YY` to the same server will likely hit its prefix cache. Therefore the EPP can build an approximate index table of the prefix caches on all the backend servers, by mimicking a similar cache eviction strategy of the model server (e.g., LRU). 
+
+Pros:
+
+* (Compared to the session affinity strategy) Broader application to most use cases and doesn't require any client integration.
+* (Compared to the consistent hashing strategy) Easy to implement and explain and is more effective.
+
+Cons:
+
+* Relies on knowledge of the cache eviction strategy of the model server, and may need careful tuning for different environments (e.g., model server with different total kv cache space may have different characteristics of cache eviction).
+* Complexity in managing cache state (eviction, memory limit)
+* An in memory cache is preferred for high performance. However, that means cache need to be rebuilt for restarts. Moreover, cache hit performance decreases with multiple active EPP replicas.
+
+## Proposal 
+
+Based on the above discussion, I propose implementing "Approximate prefix cache on the EPP" solution, which has the advantage of fast time to market, automatic prefix cache (without needing client integration), decent performance with the cost of degraded performance when sharded. 
+
+A request is broken down into N chunks of the same number of characters (we don’t necessarily need to tokenize). For each chunk we will calculate a hash based on the **content of the chunk + hash of the prefix**: `hash(chunk i) = hash(chunk i content + hash(chunk i-1))`. This gives us a nice property that if we find a match of a chunk hash, then we know all its prefix chunk hashes match as well. This is very similar to how vLLM does it.
+
+When we schedule a request `r1` with `N` chunks to a server `s1`, we update the approximate cache index table like so:
+
+```
+hash(chunk 1): append s1
+hash(chunk 2): append s1
+…
+hash(chunk N): append s1
+```
+
+This means all these N chunks are cached on server `s1`.
+
+When the EPP receives a new request `r2`, we calculate its chunk hashes, and look up the table to find a server with longest prefix matching.
+
+<img src="https://docs.google.com/drawings/d/e/2PACX-1vQ9gGbq_vrv46BZpviOUpKCuo_WCo6ANzLoAIP9lo6zrMB9kmVNk4YLKBAoGh3IsZ7mRxDu9pDqukrX/pub?w=1074&amp;h=956">
+
+[Image source](https://docs.google.com/drawings/d/1KL5DKh42Z_XzvcnejUcRymu99_HwW9y8U29IrPzRCss/edit?usp=sharing)
+
+
+## How does prefix cache affinity work with LoRA affinity and load-aware scheduling
+
+1. Prefix cache needs to be LoRA aware, as different adapters don’t share the same kv cache. Therefore when finding prefix matches, we only match for the same model/adapter.
+2. Prefix affinity needs to be aware of the server load and avoid overloading servers. We can calculate a combined weighted score of servers depending on: prefix cache hit ratio,  queue length and k-v cache utilization to achieve a good balance between prefix cache affinity and load balancing. 
+
+## Future work
+
+The main drawback of the proposed solution is the degraded performance when EPP is sharded, as the in memory cache index table loses a global view of all requests. To mitigate this issue, we can consider:
+
+* Establish a "prefix cache index reporting" protocol with model servers, and use a combination of the approximate cache index with reported indexes. This can potentially work better than a solution purely based on reported indexes, as discussed in [`Solution 3`](https://github.com/kubernetes-sigs/gateway-api-inference-extension/discussions/678).
+* When scheduling a request with low or no prefix cache in the EPP in memory index table, use the consistent hashing strategy to improve the predictability of two EPPs picking the same server, instead of random picking.
\ No newline at end of file
diff --git a/go.mod b/go.mod
index 9469d0d..9888e7b 100644
--- a/go.mod
+++ b/go.mod
@@ -15,7 +15,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
-	google.golang.org/grpc v1.72.0
+	google.golang.org/grpc v1.71.1
 	google.golang.org/protobuf v1.36.6
 	k8s.io/api v0.32.4
 	k8s.io/apiextensions-apiserver v0.32.4
@@ -25,13 +25,13 @@ require (
 	k8s.io/component-base v0.32.4
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
 	sigs.k8s.io/controller-runtime v0.20.4
-	sigs.k8s.io/gateway-api v1.2.1
-	sigs.k8s.io/structured-merge-diff/v4 v4.6.0
+	sigs.k8s.io/gateway-api v1.3.0
+	sigs.k8s.io/structured-merge-diff/v4 v4.7.0
 	sigs.k8s.io/yaml v1.4.0
 )
 
 require (
-	cel.dev/expr v0.20.0 // indirect
+	cel.dev/expr v0.19.1 // indirect
 	github.com/Masterminds/goutils v1.1.1 // indirect
 	github.com/Masterminds/semver v1.5.0 // indirect
 	github.com/Masterminds/sprig v2.22.0+incompatible // indirect
@@ -41,12 +41,12 @@ require (
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42 // indirect
+	github.com/cncf/xds/go v0.0.0-20241223141626-cff3c89139a3 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/emicklei/go-restful/v3 v3.12.0 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
 	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
-	github.com/fatih/color v1.17.0 // indirect
+	github.com/fatih/color v1.18.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
@@ -58,7 +58,7 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
-	github.com/gobuffalo/flect v1.0.2 // indirect
+	github.com/gobuffalo/flect v1.0.3 // indirect
 	github.com/goccy/go-yaml v1.11.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
@@ -91,8 +91,8 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
-	github.com/spf13/cobra v1.8.1 // indirect
-	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/cobra v1.9.1 // indirect
+	github.com/spf13/pflag v1.0.6 // indirect
 	github.com/stoewer/go-strcase v1.3.0 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
@@ -109,7 +109,7 @@ require (
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/mod v0.24.0 // indirect
 	golang.org/x/net v0.39.0 // indirect
-	golang.org/x/oauth2 v0.26.0 // indirect
+	golang.org/x/oauth2 v0.25.0 // indirect
 	golang.org/x/sync v0.13.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect
 	golang.org/x/term v0.31.0 // indirect
@@ -118,8 +118,8 @@ require (
 	golang.org/x/tools v0.31.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
@@ -129,6 +129,6 @@ require (
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.0 // indirect
-	sigs.k8s.io/controller-tools v0.16.3 // indirect
+	sigs.k8s.io/controller-tools v0.17.3 // indirect
 	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
 )
diff --git a/go.sum b/go.sum
index a47a9f9..508f9ea 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,5 @@
-cel.dev/expr v0.20.0 h1:OunBvVCfvpWlt4dN7zg3FM6TDkzOePe1+foGJ9AXeeI=
-cel.dev/expr v0.20.0/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
+cel.dev/expr v0.19.1 h1:NciYrtDRIR0lNCnH1LFJegdjspNx9fI59O7TWcua/W4=
+cel.dev/expr v0.19.1/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
 github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI=
 github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
 github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
@@ -20,9 +20,9 @@ github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK3
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42 h1:Om6kYQYDUk5wWbT0t0q6pvyM49i9XZAv9dDrkDA7gjk=
-github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
-github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/cncf/xds/go v0.0.0-20241223141626-cff3c89139a3 h1:boJj011Hh+874zpIySeApCX4GeOjPl9qhRF3QuIZq+Q=
+github.com/cncf/xds/go v0.0.0-20241223141626-cff3c89139a3/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
+github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
@@ -35,12 +35,12 @@ github.com/envoyproxy/go-control-plane/envoy v1.32.4 h1:jb83lalDRZSpPWW2Z7Mck/8k
 github.com/envoyproxy/go-control-plane/envoy v1.32.4/go.mod h1:Gzjc5k8JcJswLjAx1Zm+wSYE20UrLtt7JZMWiWQXQEw=
 github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8=
 github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU=
-github.com/evanphx/json-patch v5.7.0+incompatible h1:vgGkfT/9f8zE6tvSCe74nfpAVDQ2tG6yudJd8LBksgI=
-github.com/evanphx/json-patch v5.7.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
+github.com/evanphx/json-patch v0.5.2 h1:xVCHIVMUu1wtM/VkR9jVZ45N3FhZfYMMYGorLCR8P3k=
+github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
 github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjTM0wiaDU=
 github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM=
-github.com/fatih/color v1.17.0 h1:GlRw1BRJxkpqUCBKzKOw098ed57fEsKeNjpTe3cSjK4=
-github.com/fatih/color v1.17.0/go.mod h1:YZ7TlrGPkiz6ku9fK3TLD/pl3CpsiFyu8N92HLgmosI=
+github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
+github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
@@ -69,8 +69,8 @@ github.com/go-playground/validator/v10 v10.4.1 h1:pH2c5ADXtd66mxoE0Zm9SUhxE20r7a
 github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
 github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
-github.com/gobuffalo/flect v1.0.2 h1:eqjPGSo2WmjgY2XlpGwo2NXgL3RucAKo4k4qQMNA5sA=
-github.com/gobuffalo/flect v1.0.2/go.mod h1:A5msMlrHtLqh9umBSnvabjsMrCcCpAyzglnDvkbYKHs=
+github.com/gobuffalo/flect v1.0.3 h1:xeWBM2nui+qnVvNM4S3foBhCAL2XgPU+a7FdpelbTq4=
+github.com/gobuffalo/flect v1.0.3/go.mod h1:A5msMlrHtLqh9umBSnvabjsMrCcCpAyzglnDvkbYKHs=
 github.com/goccy/go-yaml v1.11.3 h1:B3W9IdWbvrUu2OYQGwvU1nZtvMQJPBKgBUuweJjLj6I=
 github.com/goccy/go-yaml v1.11.3/go.mod h1:wKnAMd44+9JAAnGQpWVEgBzGt3YuTaQ4uXoHvE4m7WU=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
@@ -169,10 +169,10 @@ github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoG
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
-github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
-github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y=
-github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
-github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo=
+github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0=
+github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
+github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stoewer/go-strcase v1.3.0 h1:g0eASXYtp+yvN9fK8sH94oCIk0fau9uV1/ZdJ0AVEzs=
 github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -234,8 +234,8 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=
 golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E=
-golang.org/x/oauth2 v0.26.0 h1:afQXWNNaeC4nvZ0Ed9XvCCzXM6UHJG7iCg0W4fPqSBE=
-golang.org/x/oauth2 v0.26.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
+golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -271,12 +271,12 @@ golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 h1:+cNy6SZtPcJQH3LJVLOSm
 golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
 gomodules.xyz/jsonpatch/v2 v2.4.0 h1:Ci3iUJyx9UeRx7CeFN8ARgGbkESwJK+KB9lLcWxY/Zw=
 gomodules.xyz/jsonpatch/v2 v2.4.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY=
-google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a h1:nwKuGPlUAt+aR+pcrkfFRrTU1BVrSmYyYMxYbUIVHr0=
-google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a/go.mod h1:3kWAYMk1I75K4vykHtKt2ycnOgpA6974V7bREqbsenU=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 h1:h6p3mQqrmT1XkHVTfzLdNz1u7IhINeZkz67/xTbOuWs=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
-google.golang.org/grpc v1.72.0 h1:S7UkcVa60b5AAQTaO6ZKamFp1zMZSU0fGDK2WZLbBnM=
-google.golang.org/grpc v1.72.0/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM=
+google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422 h1:GVIKPyP/kLIyVOgOnTwFOrvQaQUzOzGMCxgFUOEmm24=
+google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422/go.mod h1:b6h1vNKhxaSoEI+5jc3PJUCustfli/mRab7295pY7rw=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f h1:OxYkA3wjPsZyBylwymxSHa7ViiW1Sml4ToBrncvFehI=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f/go.mod h1:+2Yz8+CLJbIfL9z73EW45avw8Lmge3xVElCP9zEKi50=
+google.golang.org/grpc v1.71.1 h1:ffsFWr7ygTUscGPI0KKK6TLrGz0476KUvvsbqWK0rPI=
+google.golang.org/grpc v1.71.1/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
 google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
 google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -319,15 +319,15 @@ sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.0 h1:CPT0ExVicCzcp
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.0/go.mod h1:Ve9uj1L+deCXFrPOk1LpFXqTg7LCFzFso6PA48q/XZw=
 sigs.k8s.io/controller-runtime v0.20.4 h1:X3c+Odnxz+iPTRobG4tp092+CvBU9UK0t/bRf+n0DGU=
 sigs.k8s.io/controller-runtime v0.20.4/go.mod h1:xg2XB0K5ShQzAgsoujxuKN4LNXR2LfwwHsPj7Iaw+XY=
-sigs.k8s.io/controller-tools v0.16.3 h1:z48C5/d4jCVQQvtiSBL5MYyZ3EO2eFIOXrIKMgHVhFY=
-sigs.k8s.io/controller-tools v0.16.3/go.mod h1:AEj6k+w1kYpLZv2einOH3mj52ips4W/6FUjnB5tkJGs=
-sigs.k8s.io/gateway-api v1.2.1 h1:fZZ/+RyRb+Y5tGkwxFKuYuSRQHu9dZtbjenblleOLHM=
-sigs.k8s.io/gateway-api v1.2.1/go.mod h1:EpNfEXNjiYfUJypf0eZ0P5iXA9ekSGWaS1WgPaM42X0=
+sigs.k8s.io/controller-tools v0.17.3 h1:lwFPLicpBKLgIepah+c8ikRBubFW5kOQyT88r3EwfNw=
+sigs.k8s.io/controller-tools v0.17.3/go.mod h1:1ii+oXcYZkxcBXzwv3YZBlzjt1fvkrCGjVF73blosJI=
+sigs.k8s.io/gateway-api v1.3.0 h1:q6okN+/UKDATola4JY7zXzx40WO4VISk7i9DIfOvr9M=
+sigs.k8s.io/gateway-api v1.3.0/go.mod h1:d8NV8nJbaRbEKem+5IuxkL8gJGOZ+FJ+NvOIltV8gDk=
 sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 h1:/Rv+M11QRah1itp8VhT6HoVx1Ray9eB4DBr+K+/sCJ8=
 sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3/go.mod h1:18nIHnGi6636UCz6m8i4DhaJ65T6EruyzmoQqI2BVDo=
 sigs.k8s.io/randfill v0.0.0-20250304075658-069ef1bbf016 h1:kXv6kKdoEtedwuqMmkqhbkgvYKeycVbC8+iPCP9j5kQ=
 sigs.k8s.io/randfill v0.0.0-20250304075658-069ef1bbf016/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
-sigs.k8s.io/structured-merge-diff/v4 v4.6.0 h1:IUA9nvMmnKWcj5jl84xn+T5MnlZKThmUW1TdblaLVAc=
-sigs.k8s.io/structured-merge-diff/v4 v4.6.0/go.mod h1:dDy58f92j70zLsuZVuUX5Wp9vtxXpaZnkPGWeqDfCps=
+sigs.k8s.io/structured-merge-diff/v4 v4.7.0 h1:qPeWmscJcXP0snki5IYF79Z8xrl8ETFxgMd7wez1XkI=
+sigs.k8s.io/structured-merge-diff/v4 v4.7.0/go.mod h1:dDy58f92j70zLsuZVuUX5Wp9vtxXpaZnkPGWeqDfCps=
 sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
 sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=
diff --git a/pkg/epp/backend/metrics/pod_metrics.go b/pkg/epp/backend/metrics/pod_metrics.go
index 735f297..8660bc3 100644
--- a/pkg/epp/backend/metrics/pod_metrics.go
+++ b/pkg/epp/backend/metrics/pod_metrics.go
@@ -64,18 +64,22 @@ func (pm *podMetrics) GetMetrics() *Metrics {
 	return pm.metrics.Load()
 }
 
-func (pm *podMetrics) UpdatePod(in *corev1.Pod) {
-	pm.pod.Store(toInternalPod(in))
+func (pm *podMetrics) UpdatePod(pod *corev1.Pod) {
+	pm.pod.Store(toInternalPod(pod))
 }
 
-func toInternalPod(in *corev1.Pod) *backend.Pod {
+func toInternalPod(pod *corev1.Pod) *backend.Pod {
+	labels := make(map[string]string, len(pod.GetLabels()))
+	for key, value := range pod.GetLabels() {
+		labels[key] = value
+	}
 	return &backend.Pod{
 		NamespacedName: types.NamespacedName{
-			Name:      in.Name,
-			Namespace: in.Namespace,
+			Name:      pod.Name,
+			Namespace: pod.Namespace,
 		},
-		Address: in.Status.PodIP,
-		Labels:  in.Labels,
+		Address: pod.Status.PodIP,
+		Labels:  labels,
 	}
 }
 
diff --git a/pkg/epp/backend/pod.go b/pkg/epp/backend/pod.go
index 57f034b..bb7debe 100644
--- a/pkg/epp/backend/pod.go
+++ b/pkg/epp/backend/pod.go
@@ -36,12 +36,16 @@ func (p *Pod) Clone() *Pod {
 	if p == nil {
 		return nil
 	}
+	clonedLabels := make(map[string]string, len(p.Labels))
+	for key, value := range p.Labels {
+		clonedLabels[key] = value
+	}
 	return &Pod{
 		NamespacedName: types.NamespacedName{
 			Name:      p.NamespacedName.Name,
 			Namespace: p.NamespacedName.Namespace,
 		},
 		Address: p.Address,
-		Labels:  p.Labels,
+		Labels:  clonedLabels,
 	}
 }
diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go
index 22c5002..5ae4964 100644
--- a/pkg/epp/datastore/datastore.go
+++ b/pkg/epp/datastore/datastore.go
@@ -101,6 +101,11 @@ func (ds *datastore) Clear() {
 	defer ds.poolAndModelsMu.Unlock()
 	ds.pool = nil
 	ds.models = make(map[string]*v1alpha2.InferenceModel)
+	// stop all pods go routines before clearing the pods map.
+	ds.pods.Range(func(_, v any) bool {
+		v.(backendmetrics.PodMetrics).StopRefreshLoop()
+		return true
+	})
 	ds.pods.Clear()
 }
 
@@ -245,14 +250,15 @@ func (ds *datastore) PodGetAll() []backendmetrics.PodMetrics {
 
 func (ds *datastore) PodList(predicate func(backendmetrics.PodMetrics) bool) []backendmetrics.PodMetrics {
 	res := []backendmetrics.PodMetrics{}
-	fn := func(k, v any) bool {
+
+	ds.pods.Range(func(k, v any) bool {
 		pm := v.(backendmetrics.PodMetrics)
 		if predicate(pm) {
 			res = append(res, pm)
 		}
 		return true
-	}
-	ds.pods.Range(fn)
+	})
+
 	return res
 }
 
@@ -307,15 +313,14 @@ func (ds *datastore) podResyncAll(ctx context.Context, ctrlClient client.Client)
 	}
 
 	// Remove pods that don't belong to the pool or not ready any more.
-	deleteFn := func(k, v any) bool {
+	ds.pods.Range(func(k, v any) bool {
 		pm := v.(backendmetrics.PodMetrics)
 		if exist := activePods[pm.GetPod().NamespacedName.Name]; !exist {
 			logger.V(logutil.VERBOSE).Info("Removing pod", "pod", pm.GetPod())
 			ds.PodDelete(pm.GetPod().NamespacedName)
 		}
 		return true
-	}
-	ds.pods.Range(deleteFn)
+	})
 
 	return nil
 }
diff --git a/pkg/epp/handlers/request.go b/pkg/epp/handlers/request.go
index 5495c1d..1df05ce 100644
--- a/pkg/epp/handlers/request.go
+++ b/pkg/epp/handlers/request.go
@@ -18,100 +18,50 @@ package handlers
 
 import (
 	"context"
-	"encoding/json"
-	"fmt"
 	"strconv"
 	"time"
 
+	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
-	"sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
-	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+	"google.golang.org/protobuf/types/known/structpb"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
-	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
-// HandleRequestBody always returns the requestContext even in the error case, as the request context is used in error handling.
-func (s *StreamingServer) HandleRequestBody(ctx context.Context, reqCtx *RequestContext) (*RequestContext, error) {
-	logger := log.FromContext(ctx)
-
-	var requestBodyBytes []byte
-	requestBodyMap := reqCtx.Request.Body
-	// Resolve target models.
-	model, ok := requestBodyMap["model"].(string)
-	if !ok {
-		return reqCtx, errutil.Error{Code: errutil.BadRequest, Msg: "model not found in request"}
-	}
-	prompt, ok := requestBodyMap["prompt"].(string)
-	if !ok {
-		return reqCtx, errutil.Error{Code: errutil.BadRequest, Msg: "prompt not found in request"}
-	}
-
-	modelName := model
+func (s *StreamingServer) HandleRequestHeaders(ctx context.Context, reqCtx *RequestContext, req *extProcPb.ProcessingRequest_RequestHeaders) error {
+	reqCtx.RequestReceivedTimestamp = time.Now()
 
-	// NOTE: The nil checking for the modelObject means that we DO allow passthrough currently.
-	// This might be a security risk in the future where adapters not registered in the InferenceModel
-	// are able to be requested by using their distinct name.
-	modelObj := s.datastore.ModelGet(model)
-	if modelObj == nil {
-		return reqCtx, errutil.Error{Code: errutil.BadConfiguration, Msg: fmt.Sprintf("error finding a model object in InferenceModel for input %v", model)}
-	}
-	if len(modelObj.Spec.TargetModels) > 0 {
-		modelName = RandomWeightedDraw(logger, modelObj, 0)
-		if modelName == "" {
-			return reqCtx, errutil.Error{Code: errutil.BadConfiguration, Msg: fmt.Sprintf("error getting target model name for model %v", modelObj.Name)}
+	// an EoS in the request headers means this request has no body or trailers.
+	if req.RequestHeaders.EndOfStream {
+		// We will route this request to a random pod as this is assumed to just be a GET
+		// More context: https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/526
+		// The above PR will address endpoint admission, but currently any request without a body will be
+		// routed to a random upstream pod.
+		pod := s.director.GetRandomPod()
+		if pod == nil {
+			return errutil.Error{Code: errutil.Internal, Msg: "no pods available in datastore"}
 		}
+		pool, err := s.datastore.PoolGet()
+		if err != nil {
+			return err
+		}
+		reqCtx.TargetEndpoint = pod.Address + ":" + strconv.Itoa(int(pool.Spec.TargetPortNumber))
+		reqCtx.RequestSize = 0
+		reqCtx.reqHeaderResp = s.generateRequestHeaderResponse(reqCtx)
+		return nil
 	}
-	llmReq := &schedulingtypes.LLMRequest{
-		Model:               model,
-		ResolvedTargetModel: modelName,
-		Critical:            modelObj.Spec.Criticality != nil && *modelObj.Spec.Criticality == v1alpha2.Critical,
-		Prompt:              prompt,
-		Headers:             reqCtx.Request.Headers,
-	}
-	logger.V(logutil.DEBUG).Info("LLM request assembled", "request", llmReq)
-
-	var err error
-	// Update target models in the body.
-	if llmReq.Model != llmReq.ResolvedTargetModel {
-		requestBodyMap["model"] = llmReq.ResolvedTargetModel
-	}
-
-	requestBodyBytes, err = json.Marshal(requestBodyMap)
-	if err != nil {
-		logger.V(logutil.DEFAULT).Error(err, "Error marshaling request body")
-		return reqCtx, errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("error marshaling request body: %v", err)}
-	}
-
-	res, err := s.scheduler.Schedule(ctx, llmReq)
-	if err != nil {
-		return reqCtx, errutil.Error{Code: errutil.InferencePoolResourceExhausted, Msg: fmt.Errorf("failed to find target pod: %w", err).Error()}
-	}
-	targetPod := res.TargetPod.GetPod()
 
-	// Insert target endpoint to instruct Envoy to route requests to the specified target pod.
-	// Attach the port number
-	pool, err := s.datastore.PoolGet()
-	if err != nil {
-		return reqCtx, err
+	for _, header := range req.RequestHeaders.Headers.Headers {
+		if header.RawValue != nil {
+			reqCtx.Request.Headers[header.Key] = string(header.RawValue)
+		} else {
+			reqCtx.Request.Headers[header.Key] = header.Value
+		}
 	}
-	endpoint := targetPod.Address + ":" + strconv.Itoa(int(pool.Spec.TargetPortNumber))
-
-	logger.V(logutil.DEFAULT).Info("Request handled",
-		"model", llmReq.Model, "targetModel", llmReq.ResolvedTargetModel, "endpoint", targetPod)
-
-	reqCtx.Model = llmReq.Model
-	reqCtx.ResolvedTargetModel = llmReq.ResolvedTargetModel
-	reqCtx.RequestSize = len(requestBodyBytes)
-	reqCtx.TargetPod = targetPod.NamespacedName.String()
-	reqCtx.TargetEndpoint = endpoint
-
-	s.populateRequestHeaderResponse(reqCtx, endpoint, len(requestBodyBytes))
+	return nil
+}
 
-	reqCtx.reqBodyResp = &extProcPb.ProcessingResponse{
-		// The Endpoint Picker supports two approaches to communicating the target endpoint, as a request header
-		// and as an unstructure ext-proc response metadata key/value pair. This enables different integration
-		// options for gateway providers.
+func (s *StreamingServer) generateRequestBodyResponse(requestBodyBytes []byte) *extProcPb.ProcessingResponse {
+	return &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_RequestBody{
 			RequestBody: &extProcPb.BodyResponse{
 				Response: &extProcPb.CommonResponse{
@@ -127,37 +77,82 @@ func (s *StreamingServer) HandleRequestBody(ctx context.Context, reqCtx *Request
 			},
 		},
 	}
-	return reqCtx, nil
 }
 
-func (s *StreamingServer) HandleRequestHeaders(ctx context.Context, reqCtx *RequestContext, req *extProcPb.ProcessingRequest_RequestHeaders) error {
-	reqCtx.RequestReceivedTimestamp = time.Now()
+func (s *StreamingServer) generateRequestHeaderResponse(reqCtx *RequestContext) *extProcPb.ProcessingResponse {
+	// The Endpoint Picker supports two approaches to communicating the target endpoint, as a request header
+	// and as an unstructure ext-proc response metadata key/value pair. This enables different integration
+	// options for gateway providers.
+	return &extProcPb.ProcessingResponse{
+		Response: &extProcPb.ProcessingResponse_RequestHeaders{
+			RequestHeaders: &extProcPb.HeadersResponse{
+				Response: &extProcPb.CommonResponse{
+					ClearRouteCache: true,
+					HeaderMutation: &extProcPb.HeaderMutation{
+						SetHeaders: s.generateHeaders(reqCtx),
+					},
+				},
+			},
+		},
+		DynamicMetadata: s.generateMetadata(reqCtx.TargetEndpoint),
+	}
+}
 
-	// an EoS in the request headers means this request has no body or trailers.
-	if req.RequestHeaders.EndOfStream {
-		// We will route this request to a random pod as this is assumed to just be a GET
-		// More context: https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/526
-		// The above PR will address endpoint admission, but currently any request without a body will be
-		// routed to a random upstream pod.
-		pod := GetRandomPod(s.datastore)
-		if pod == nil {
-			return errutil.Error{Code: errutil.Internal, Msg: "no pods available in datastore"}
-		}
-		pool, err := s.datastore.PoolGet()
-		if err != nil {
-			return err
-		}
-		endpoint := pod.Address + ":" + strconv.Itoa(int(pool.Spec.TargetPortNumber))
-		s.populateRequestHeaderResponse(reqCtx, endpoint, 0)
-		return nil
+func (s *StreamingServer) generateHeaders(reqCtx *RequestContext) []*configPb.HeaderValueOption {
+	// can likely refactor these two bespoke headers to be updated in PostDispatch, to centralize logic.
+	headers := []*configPb.HeaderValueOption{
+		{
+			Header: &configPb.HeaderValue{
+				Key:      s.destinationEndpointHintKey,
+				RawValue: []byte(reqCtx.TargetEndpoint),
+			},
+		},
+	}
+	if reqCtx.RequestSize > 0 {
+		// We need to update the content length header if the body is mutated, see Envoy doc:
+		// https://www.envoyproxy.io/docs/envoy/latest/api-v3/extensions/filters/http/ext_proc/v3/processing_mode.proto
+		headers = append(headers, &configPb.HeaderValueOption{
+			Header: &configPb.HeaderValue{
+				Key:      "Content-Length",
+				RawValue: []byte(strconv.Itoa(reqCtx.RequestSize)),
+			},
+		})
 	}
 
-	for _, header := range req.RequestHeaders.Headers.Headers {
-		if header.RawValue != nil {
-			reqCtx.Request.Headers[header.Key] = string(header.RawValue)
-		} else {
-			reqCtx.Request.Headers[header.Key] = header.Value
+	// include all headers
+	for key, value := range reqCtx.Request.Headers {
+		headers = append(headers, &configPb.HeaderValueOption{
+			Header: &configPb.HeaderValue{
+				Key:      key,
+				RawValue: []byte(value),
+			},
+		})
+	}
+	return headers
+}
+
+func (s *StreamingServer) generateMetadata(endpoint string) *structpb.Struct {
+	targetEndpointValue := &structpb.Struct{
+		Fields: map[string]*structpb.Value{
+			s.destinationEndpointHintKey: {
+				Kind: &structpb.Value_StringValue{
+					StringValue: endpoint,
+				},
+			},
+		},
+	}
+	dynamicMetadata := targetEndpointValue
+	if s.destinationEndpointHintMetadataNamespace != "" {
+		// If a namespace is defined, wrap the selected endpoint with that.
+		dynamicMetadata = &structpb.Struct{
+			Fields: map[string]*structpb.Value{
+				s.destinationEndpointHintMetadataNamespace: {
+					Kind: &structpb.Value_StructValue{
+						StructValue: targetEndpointValue,
+					},
+				},
+			},
 		}
 	}
-	return nil
+	return dynamicMetadata
 }
diff --git a/pkg/epp/handlers/response.go b/pkg/epp/handlers/response.go
index 04c7a5e..bbc46c9 100644
--- a/pkg/epp/handlers/response.go
+++ b/pkg/epp/handlers/response.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"strings"
 
+	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
@@ -98,6 +99,58 @@ func (s *StreamingServer) HandleResponseBodyModelStreaming(
 	}
 }
 
+func (s *StreamingServer) HandleResponseHeaders(ctx context.Context, reqCtx *RequestContext, resp *extProcPb.ProcessingRequest_ResponseHeaders) (*RequestContext, error) {
+	for _, header := range resp.ResponseHeaders.Headers.Headers {
+		if header.RawValue != nil {
+			reqCtx.Response.Headers[header.Key] = string(header.RawValue)
+		} else {
+			reqCtx.Response.Headers[header.Key] = header.Value
+		}
+	}
+
+	reqCtx, err := s.director.HandleResponse(ctx, reqCtx)
+
+	return reqCtx, err
+}
+
+func (s *StreamingServer) generateResponseHeaderResponse(reqCtx *RequestContext) *extProcPb.ProcessingResponse {
+	return &extProcPb.ProcessingResponse{
+		Response: &extProcPb.ProcessingResponse_ResponseHeaders{
+			ResponseHeaders: &extProcPb.HeadersResponse{
+				Response: &extProcPb.CommonResponse{
+					HeaderMutation: &extProcPb.HeaderMutation{
+						SetHeaders: s.generateResponseHeaders(reqCtx),
+					},
+				},
+			},
+		},
+	}
+}
+
+func (s *StreamingServer) generateResponseHeaders(reqCtx *RequestContext) []*configPb.HeaderValueOption {
+	// can likely refactor these two bespoke headers to be updated in PostDispatch, to centralize logic.
+	headers := []*configPb.HeaderValueOption{
+		{
+			Header: &configPb.HeaderValue{
+				// This is for debugging purpose only.
+				Key:      "x-went-into-resp-headers",
+				RawValue: []byte("true"),
+			},
+		},
+	}
+
+	// include all headers
+	for key, value := range reqCtx.Response.Headers {
+		headers = append(headers, &configPb.HeaderValueOption{
+			Header: &configPb.HeaderValue{
+				Key:      key,
+				RawValue: []byte(value),
+			},
+		})
+	}
+	return headers
+}
+
 // Example message if "stream_options": {"include_usage": "true"} is included in the request:
 // data: {"id":"...","object":"text_completion","created":1739400043,"model":"food-review-0","choices":[],
 // "usage":{"prompt_tokens":7,"total_tokens":17,"completion_tokens":10}}
@@ -112,8 +165,8 @@ func (s *StreamingServer) HandleResponseBodyModelStreaming(
 func parseRespForUsage(
 	ctx context.Context,
 	responseText string,
-) Response {
-	response := Response{}
+) ResponseBody {
+	response := ResponseBody{}
 	logger := log.FromContext(ctx)
 
 	lines := strings.Split(responseText, "\n")
@@ -136,7 +189,7 @@ func parseRespForUsage(
 	return response
 }
 
-type Response struct {
+type ResponseBody struct {
 	Usage Usage `json:"usage"`
 }
 
diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go
index 711d8ce..be85ba6 100644
--- a/pkg/epp/handlers/server.go
+++ b/pkg/epp/handlers/server.go
@@ -20,56 +20,59 @@ import (
 	"context"
 	"encoding/json"
 	"io"
-	"math/rand"
-	"strconv"
 	"strings"
 	"time"
 
-	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	envoyTypePb "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/go-logr/logr"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/types/known/structpb"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
-	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
 )
 
-func NewStreamingServer(scheduler Scheduler, destinationEndpointHintMetadataNamespace, destinationEndpointHintKey string, datastore datastore.Datastore) *StreamingServer {
+func NewStreamingServer(destinationEndpointHintMetadataNamespace, destinationEndpointHintKey string, datastore Datastore, director Director) *StreamingServer {
 	return &StreamingServer{
-		scheduler:                                scheduler,
 		destinationEndpointHintMetadataNamespace: destinationEndpointHintMetadataNamespace,
 		destinationEndpointHintKey:               destinationEndpointHintKey,
+		director:                                 director,
 		datastore:                                datastore,
 	}
 }
 
+type Director interface {
+	HandleRequest(ctx context.Context, reqCtx *RequestContext) (*RequestContext, error)
+	HandleResponse(ctx context.Context, reqCtx *RequestContext) (*RequestContext, error)
+	GetRandomPod() *backend.Pod
+}
+
+type Datastore interface {
+	PoolGet() (*v1alpha2.InferencePool, error)
+}
+
 // Server implements the Envoy external processing server.
 // https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/ext_proc/v3/external_processor.proto
 type StreamingServer struct {
-	scheduler Scheduler
 	// The key of the header to specify the target pod address. This value needs to match Envoy
 	// configuration.
 	destinationEndpointHintKey string
 	// The key acting as the outer namespace struct in the metadata extproc response to communicate
 	// back the picked endpoints.
 	destinationEndpointHintMetadataNamespace string
-	datastore                                datastore.Datastore
-}
-
-type Scheduler interface {
-	Schedule(ctx context.Context, b *schedulingtypes.LLMRequest) (result *schedulingtypes.Result, err error)
+	datastore                                Datastore
+	director                                 Director
 }
 
 // RequestContext stores context information during the life time of an HTTP request.
+// TODO: The requestContext is gathering a ton of fields. A future refactor needs to tease these fields apart.
+// Specifically, there are fields related to the ext-proc protocol, and then fields related to the lifecycle of the request.
+// We should split these apart as this monolithic object exposes too much data to too many layers.
 type RequestContext struct {
 	TargetPod                 string
 	TargetEndpoint            string
@@ -88,6 +91,8 @@ type RequestContext struct {
 	RequestState         StreamRequestState
 	modelServerStreaming bool
 
+	Response *Response
+
 	reqHeaderResp  *extProcPb.ProcessingResponse
 	reqBodyResp    *extProcPb.ProcessingResponse
 	reqTrailerResp *extProcPb.ProcessingResponse
@@ -101,6 +106,9 @@ type Request struct {
 	Headers map[string]string
 	Body    map[string]interface{}
 }
+type Response struct {
+	Headers map[string]string
+}
 type StreamRequestState int
 
 const (
@@ -128,6 +136,9 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 			Headers: make(map[string]string),
 			Body:    make(map[string]interface{}),
 		},
+		Response: &Response{
+			Headers: make(map[string]string),
+		},
 	}
 
 	var body []byte
@@ -192,13 +203,24 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				// Body stream complete. Allocate empty slice for response to use.
 				body = []byte{}
 
-				reqCtx, err = s.HandleRequestBody(ctx, reqCtx)
+				reqCtx, err = s.director.HandleRequest(ctx, reqCtx)
 				if err != nil {
-					logger.V(logutil.DEFAULT).Error(err, "Error handling body")
-				} else {
-					metrics.RecordRequestCounter(reqCtx.Model, reqCtx.ResolvedTargetModel)
-					metrics.RecordRequestSizes(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.RequestSize)
+					logger.V(logutil.DEFAULT).Error(err, "Error handling request")
+					break
 				}
+
+				// Populate the ExtProc protocol responses for the request body.
+				requestBodyBytes, err := json.Marshal(reqCtx.Request.Body)
+				if err != nil {
+					logger.V(logutil.DEFAULT).Error(err, "Error marshalling request body")
+					break
+				}
+				reqCtx.RequestSize = len(requestBodyBytes)
+				reqCtx.reqHeaderResp = s.generateRequestHeaderResponse(reqCtx)
+				reqCtx.reqBodyResp = s.generateRequestBodyResponse(requestBodyBytes)
+
+				metrics.RecordRequestCounter(reqCtx.Model, reqCtx.ResolvedTargetModel)
+				metrics.RecordRequestSizes(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.RequestSize)
 			}
 		case *extProcPb.ProcessingRequest_RequestTrailers:
 			// This is currently unused.
@@ -215,25 +237,13 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				}
 			}
 			reqCtx.RequestState = ResponseRecieved
-			reqCtx.respHeaderResp = &extProcPb.ProcessingResponse{
-				Response: &extProcPb.ProcessingResponse_ResponseHeaders{
-					ResponseHeaders: &extProcPb.HeadersResponse{
-						Response: &extProcPb.CommonResponse{
-							HeaderMutation: &extProcPb.HeaderMutation{
-								SetHeaders: []*configPb.HeaderValueOption{
-									{
-										Header: &configPb.HeaderValue{
-											// This is for debugging purpose only.
-											Key:      "x-went-into-resp-headers",
-											RawValue: []byte("true"),
-										},
-									},
-								},
-							},
-						},
-					},
-				},
+
+			var responseErr error
+			reqCtx, responseErr = s.HandleResponseHeaders(ctx, reqCtx, v)
+			if responseErr != nil {
+				logger.V(logutil.DEFAULT).Error(responseErr, "Failed to process response headers", "request", req)
 			}
+			reqCtx.respHeaderResp = s.generateResponseHeaderResponse(reqCtx)
 
 		case *extProcPb.ProcessingRequest_ResponseBody:
 			if reqCtx.modelServerStreaming {
@@ -375,105 +385,6 @@ func (r *RequestContext) updateStateAndSendIfNeeded(srv extProcPb.ExternalProces
 	return nil
 }
 
-func (s *StreamingServer) populateRequestHeaderResponse(reqCtx *RequestContext, endpoint string, requestBodyLength int) {
-	headers := []*configPb.HeaderValueOption{
-		{
-			Header: &configPb.HeaderValue{
-				Key:      s.destinationEndpointHintKey,
-				RawValue: []byte(endpoint),
-			},
-		},
-	}
-	if requestBodyLength > 0 {
-		// We need to update the content length header if the body is mutated, see Envoy doc:
-		// https://www.envoyproxy.io/docs/envoy/latest/api-v3/extensions/filters/http/ext_proc/v3/processing_mode.proto
-		headers = append(headers, &configPb.HeaderValueOption{
-			Header: &configPb.HeaderValue{
-				Key:      "Content-Length",
-				RawValue: []byte(strconv.Itoa(requestBodyLength)),
-			},
-		})
-	}
-
-	targetEndpointValue := &structpb.Struct{
-		Fields: map[string]*structpb.Value{
-			s.destinationEndpointHintKey: {
-				Kind: &structpb.Value_StringValue{
-					StringValue: endpoint,
-				},
-			},
-		},
-	}
-	dynamicMetadata := targetEndpointValue
-	if s.destinationEndpointHintMetadataNamespace != "" {
-		// If a namespace is defined, wrap the selected endpoint with that.
-		dynamicMetadata = &structpb.Struct{
-			Fields: map[string]*structpb.Value{
-				s.destinationEndpointHintMetadataNamespace: {
-					Kind: &structpb.Value_StructValue{
-						StructValue: targetEndpointValue,
-					},
-				},
-			},
-		}
-	}
-
-	reqCtx.reqHeaderResp = &extProcPb.ProcessingResponse{
-		Response: &extProcPb.ProcessingResponse_RequestHeaders{
-			RequestHeaders: &extProcPb.HeadersResponse{
-				Response: &extProcPb.CommonResponse{
-					ClearRouteCache: true,
-					HeaderMutation: &extProcPb.HeaderMutation{
-						SetHeaders: headers,
-					},
-				},
-			},
-		},
-		DynamicMetadata: dynamicMetadata,
-	}
-}
-
-func RandomWeightedDraw(logger logr.Logger, model *v1alpha2.InferenceModel, seed int64) string {
-	// TODO: after we are down to 1 server implementation, make these methods a part of the struct
-	// and handle random seeding on the struct.
-	source := rand.NewSource(rand.Int63())
-	if seed > 0 {
-		source = rand.NewSource(seed)
-	}
-	r := rand.New(source)
-
-	// all the weight values are nil, then we should return random model name
-	if model.Spec.TargetModels[0].Weight == nil {
-		index := r.Int31n(int32(len(model.Spec.TargetModels)))
-		return model.Spec.TargetModels[index].Name
-	}
-
-	var weights int32
-	for _, model := range model.Spec.TargetModels {
-		weights += *model.Weight
-	}
-	logger.V(logutil.TRACE).Info("Weights for model computed", "model", model.Name, "weights", weights)
-	randomVal := r.Int31n(weights)
-	// TODO: optimize this without using loop
-	for _, model := range model.Spec.TargetModels {
-		if randomVal < *model.Weight {
-			return model.Name
-		}
-		randomVal -= *model.Weight
-	}
-	return ""
-}
-
-func GetRandomPod(ds datastore.Datastore) *backend.Pod {
-	pods := ds.PodGetAll()
-	if len(pods) == 0 {
-		return nil
-	}
-	number := rand.Intn(len(pods))
-	pod := pods[number]
-	return pod.GetPod()
-}
-
 func BuildErrResponse(err error) (*extProcPb.ProcessingResponse, error) {
 	var resp *extProcPb.ProcessingResponse
 
diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go
index 6cc0cdb..84f0f1f 100644
--- a/pkg/epp/metrics/metrics.go
+++ b/pkg/epp/metrics/metrics.go
@@ -209,6 +209,40 @@ var (
 		[]string{"plugin_type", "plugin_name"},
 	)
 
+	// Prefix indexer Metrics
+	PrefixCacheSize = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
+			Subsystem:      InferenceExtension,
+			Name:           "prefix_indexer_size",
+			Help:           "Size of the prefix indexer.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{},
+	)
+
+	PrefixCacheHitRatio = compbasemetrics.NewHistogramVec(
+		&compbasemetrics.HistogramOpts{
+			Subsystem: InferenceExtension,
+			Name:      "prefix_indexer_hit_ratio",
+			Help:      "Ratio of prefix length matched to total prefix length in the cache lookup.",
+			// Buckets from 0.0 to 1.0 in increments
+			Buckets:        []float64{0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0},
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{},
+	)
+
+	PrefixCacheHitLength = compbasemetrics.NewHistogramVec(
+		&compbasemetrics.HistogramOpts{
+			Subsystem:      InferenceExtension,
+			Name:           "prefix_indexer_hit_bytes",
+			Help:           "Length of the prefix match in number of bytes in the cache lookup.",
+			Buckets:        []float64{0, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768, 65536},
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{},
+	)
+
 	// Info Metrics
 	InferenceExtensionInfo = compbasemetrics.NewGaugeVec(
 		&compbasemetrics.GaugeOpts{
@@ -244,6 +278,10 @@ func Register() {
 		legacyregistry.MustRegister(SchedulerE2ELatency)
 
 		legacyregistry.MustRegister(InferenceExtensionInfo)
+
+		legacyregistry.MustRegister(PrefixCacheSize)
+		legacyregistry.MustRegister(PrefixCacheHitRatio)
+		legacyregistry.MustRegister(PrefixCacheHitLength)
 	})
 }
 
@@ -352,6 +390,24 @@ func RecordSchedulerE2ELatency(duration time.Duration) {
 	SchedulerE2ELatency.WithLabelValues().Observe(duration.Seconds())
 }
 
+// RecordPrefixCacheSize records the size of the prefix indexer in megabytes.
+func RecordPrefixCacheSize(size int64) {
+	PrefixCacheSize.WithLabelValues().Set(float64(size))
+}
+
+// RecordPrefixCacheMatch records both the hit ratio and hit length for a prefix indexer match.
+// matchedLength is the number of characters that matched, and totalLength is the total prefix length.
+func RecordPrefixCacheMatch(matchedLength, totalLength int) {
+	// Record the hit length metric
+	PrefixCacheHitLength.WithLabelValues().Observe(float64(matchedLength))
+
+	// Record the hit ratio metric if totalLength is positive
+	if totalLength > 0 {
+		ratio := float64(matchedLength) / float64(totalLength)
+		PrefixCacheHitRatio.WithLabelValues().Observe(ratio)
+	}
+}
+
 func RecordInferenceExtensionInfo() {
 	if CommitSHA != "" {
 		InferenceExtensionInfo.WithLabelValues(CommitSHA).Set(1)
diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go
new file mode 100644
index 0000000..bfcf2ec
--- /dev/null
+++ b/pkg/epp/requestcontrol/director.go
@@ -0,0 +1,196 @@
+/*
+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"
+	"fmt"
+	"math/rand"
+	"strconv"
+
+	"github.com/go-logr/logr"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
+	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
+)
+
+type Scheduler interface {
+	Schedule(ctx context.Context, b *schedulingtypes.LLMRequest) (result *schedulingtypes.Result, err error)
+	OnResponse(ctx context.Context, resp *schedulingtypes.LLMResponse, targetPodName string)
+}
+
+type Director struct {
+	datastore datastore.Datastore
+	scheduler Scheduler
+}
+
+func NewDirector(datastore datastore.Datastore, scheduler Scheduler) *Director {
+	return &Director{
+		datastore: datastore,
+		scheduler: scheduler,
+	}
+}
+
+// HandleRequest always returns the requestContext even in the error case, as the request context is used in error handling.
+func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestContext) (*handlers.RequestContext, error) {
+	logger := log.FromContext(ctx)
+
+	// Resolve target models.
+	var ok bool
+	requestBodyMap := reqCtx.Request.Body
+	reqCtx.Model, ok = requestBodyMap["model"].(string)
+	if !ok {
+		return reqCtx, errutil.Error{Code: errutil.BadRequest, Msg: "model not found in request"}
+	}
+	prompt, ok := requestBodyMap["prompt"].(string)
+	if !ok {
+		return reqCtx, errutil.Error{Code: errutil.BadRequest, Msg: "prompt not found in request"}
+	}
+
+	// NOTE: The nil checking for the modelObject means that we DO allow passthrough currently.
+	// This might be a security risk in the future where adapters not registered in the InferenceModel
+	// are able to be requested by using their distinct name.
+	modelObj := d.datastore.ModelGet(reqCtx.Model)
+	if modelObj == nil {
+		return reqCtx, errutil.Error{Code: errutil.BadConfiguration, Msg: fmt.Sprintf("error finding a model object in InferenceModel for input %v", reqCtx.Model)}
+	}
+
+	reqCtx.ResolvedTargetModel = reqCtx.Model
+	if len(modelObj.Spec.TargetModels) > 0 {
+		reqCtx.ResolvedTargetModel = RandomWeightedDraw(logger, modelObj, 0)
+		if reqCtx.ResolvedTargetModel == "" {
+			return reqCtx, errutil.Error{Code: errutil.BadConfiguration, Msg: fmt.Sprintf("error getting target model name for model %v", modelObj.Name)}
+		}
+		reqCtx.Request.Body["model"] = reqCtx.ResolvedTargetModel // Update target model in the body.
+	}
+
+	llmReq := &schedulingtypes.LLMRequest{
+		TargetModel: reqCtx.ResolvedTargetModel,
+		RequestId:   reqCtx.Request.Headers[requtil.RequestIdHeaderKey],
+		Critical:    modelObj.Spec.Criticality != nil && *modelObj.Spec.Criticality == v1alpha2.Critical,
+		Prompt:      prompt,
+		Headers:     reqCtx.Request.Headers,
+	}
+	logger.V(logutil.DEBUG).Info("LLM request assembled", "request", llmReq)
+	results, err := d.Dispatch(ctx, llmReq)
+	if err != nil {
+		return reqCtx, err
+	}
+
+	// Insert target endpoint to instruct Envoy to route requests to the specified target pod.
+	// Attach the port number
+	reqCtx, err = d.PostDispatch(ctx, reqCtx, results)
+	if err != nil {
+		return reqCtx, err
+	}
+
+	return reqCtx, nil
+}
+
+// Dispatch runs one or many scheduling cycles.
+func (d *Director) Dispatch(ctx context.Context, llmReq *schedulingtypes.LLMRequest) ([]*schedulingtypes.Result, error) {
+	var err error
+	res, err := d.scheduler.Schedule(ctx, llmReq)
+	if err != nil {
+		return nil, errutil.Error{Code: errutil.InferencePoolResourceExhausted, Msg: fmt.Errorf("failed to find target pod: %w", err).Error()}
+	}
+
+	return []*schedulingtypes.Result{res}, nil
+}
+
+func (d *Director) PostDispatch(ctx context.Context, reqCtx *handlers.RequestContext, results []*schedulingtypes.Result) (*handlers.RequestContext, error) {
+	logger := log.FromContext(ctx)
+	// currently only get a single result. Will refactor to pluggably implement the PostSchedule
+	if len(results) == 0 {
+		return reqCtx, errutil.Error{Code: errutil.Internal, Msg: "results must be greater than zero"}
+	}
+	targetPod := results[0].TargetPod.GetPod()
+
+	pool, err := d.datastore.PoolGet()
+	if err != nil {
+		return reqCtx, err
+	}
+
+	endpoint := targetPod.Address + ":" + strconv.Itoa(int(pool.Spec.TargetPortNumber))
+	logger.V(logutil.DEFAULT).Info("Request handled", "model", reqCtx.Model, "targetModel", reqCtx.ResolvedTargetModel, "endpoint", targetPod)
+
+	reqCtx.TargetPod = targetPod.NamespacedName.String()
+	reqCtx.TargetEndpoint = endpoint
+
+	return reqCtx, nil
+}
+
+func (d *Director) HandleResponse(ctx context.Context, reqCtx *handlers.RequestContext) (*handlers.RequestContext, error) {
+	logger := log.FromContext(ctx)
+
+	llmResp := &schedulingtypes.LLMResponse{
+		RequestId: reqCtx.Request.Headers[requtil.RequestIdHeaderKey],
+		Headers:   reqCtx.Response.Headers,
+	}
+	logger.V(logutil.DEBUG).Info("LLM response assembled", "response", llmResp)
+
+	d.scheduler.OnResponse(ctx, llmResp, reqCtx.TargetPod)
+
+	return reqCtx, nil
+}
+
+func (d *Director) GetRandomPod() *backend.Pod {
+	pods := d.datastore.PodGetAll()
+	if len(pods) == 0 {
+		return nil
+	}
+	number := rand.Intn(len(pods))
+	pod := pods[number]
+	return pod.GetPod()
+}
+
+func RandomWeightedDraw(logger logr.Logger, model *v1alpha2.InferenceModel, seed int64) string {
+	// TODO: after we are down to 1 server implementation, make these methods a part of the struct
+	// and handle random seeding on the struct.
+	source := rand.NewSource(rand.Int63())
+	if seed > 0 {
+		source = rand.NewSource(seed)
+	}
+	r := rand.New(source)
+
+	// all the weight values are nil, then we should return random model name
+	if model.Spec.TargetModels[0].Weight == nil {
+		index := r.Int31n(int32(len(model.Spec.TargetModels)))
+		return model.Spec.TargetModels[index].Name
+	}
+
+	var weights int32
+	for _, model := range model.Spec.TargetModels {
+		weights += *model.Weight
+	}
+	logger.V(logutil.TRACE).Info("Weights for model computed", "model", model.Name, "weights", weights)
+	randomVal := r.Int31n(weights)
+	// TODO: optimize this without using loop
+	for _, model := range model.Spec.TargetModels {
+		if randomVal < *model.Weight {
+			return model.Name
+		}
+		randomVal -= *model.Weight
+	}
+	return ""
+}
diff --git a/pkg/epp/scheduling/config.go b/pkg/epp/scheduling/config.go
index c02d9f5..0292289 100644
--- a/pkg/epp/scheduling/config.go
+++ b/pkg/epp/scheduling/config.go
@@ -16,18 +16,26 @@ limitations under the License.
 
 package scheduling
 
-import "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/prefix"
+)
 
 // NewSchedulerConfig creates a new SchedulerConfig object with the given plugins.
 func NewSchedulerConfig(preSchedulePlugins []plugins.PreSchedule, filters []plugins.Filter, scorers map[plugins.Scorer]int,
-	picker plugins.Picker, postSchedulePlugins []plugins.PostSchedule) *SchedulerConfig {
-	return &SchedulerConfig{
+	picker plugins.Picker, postSchedulePlugins []plugins.PostSchedule, postResponsePlugins []plugins.PostResponse, opts ...ConfigOption) *SchedulerConfig {
+	config := &SchedulerConfig{
 		preSchedulePlugins:  preSchedulePlugins,
 		filters:             filters,
 		scorers:             scorers,
 		picker:              picker,
 		postSchedulePlugins: postSchedulePlugins,
+		postResponsePlugins: postResponsePlugins,
 	}
+	for _, opt := range opts {
+		opt(config)
+	}
+	return config
 }
 
 // SchedulerConfig provides a configuration for the scheduler which influence routing decisions.
@@ -37,18 +45,18 @@ type SchedulerConfig struct {
 	scorers             map[plugins.Scorer]int // map from scorer to weight
 	picker              plugins.Picker
 	postSchedulePlugins []plugins.PostSchedule
+	postResponsePlugins []plugins.PostResponse
 }
 
-var defPlugin = &defaultPlugin{}
-
-// When the scheduler is initialized with NewScheduler function, this config will be used as default.
-// it's possible to call NewSchedulerWithConfig to pass a different argument.
+type ConfigOption func(*SchedulerConfig)
 
-// For build time plugins changes, it's recommended to change the defaultConfig variable in this file.
-var defaultConfig = &SchedulerConfig{
-	preSchedulePlugins:  []plugins.PreSchedule{},
-	filters:             []plugins.Filter{defPlugin},
-	scorers:             map[plugins.Scorer]int{},
-	picker:              defPlugin,
-	postSchedulePlugins: []plugins.PostSchedule{},
+// TODO(https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/813): Replace this
+// with a more generic way to add plugins.
+func AddPrefixPlugin(prefixConfig prefix.Config, weight int) ConfigOption {
+	return func(cfg *SchedulerConfig) {
+		prefixPlugin := prefix.New(prefixConfig)
+		cfg.preSchedulePlugins = append(cfg.preSchedulePlugins, prefixPlugin)
+		cfg.postSchedulePlugins = append(cfg.postSchedulePlugins, prefixPlugin)
+		cfg.scorers[prefixPlugin] = weight
+	}
 }
diff --git a/pkg/epp/scheduling/plugins/filter/decision_tree_filter.go b/pkg/epp/scheduling/plugins/filter/decision_tree_filter.go
new file mode 100644
index 0000000..399780b
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/filter/decision_tree_filter.go
@@ -0,0 +1,81 @@
+/*
+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 filter
+
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+// DecisionTreeFilter applies current fitler, and then recursively applies next filters
+// depending success or failure of the current filter.
+// It can be used to construct a flow chart algorithm.
+type DecisionTreeFilter struct {
+	Current plugins.Filter
+	// NextOnSuccess filter will be applied after successfully applying the current filter.
+	// The filtered results will be passed to the next filter.
+	NextOnSuccess plugins.Filter
+	// NextOnFailure filter will be applied if current filter results in no pods.
+	// The original input will be passed to the next filter.
+	NextOnFailure plugins.Filter
+	// NextOnSuccessOrFailure is a convenience field to configure the next filter regardless of the
+	// success or failure of the current filter.
+	// NOTE: When using NextOnSuccessOrFailure, both nextOnSuccess and nextOnFailure SHOULD be nil.
+	// However if that's not the case, nextOnSuccess and nextOnFailure will be used, instead of
+	// NextOnSuccessOrFailure, in the success and failure scenarios, respectively.
+	NextOnSuccessOrFailure plugins.Filter
+}
+
+// Name returns the name of the filter.
+func (f *DecisionTreeFilter) Name() string {
+	if f == nil {
+		return "nil"
+	}
+	return f.Current.Name()
+}
+
+// Filter filters out pods that doesn't meet the filter criteria.
+func (f *DecisionTreeFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	loggerTrace := ctx.Logger.V(logutil.TRACE)
+	filteredPod := f.Current.Filter(ctx, pods)
+
+	next := f.NextOnSuccessOrFailure
+	if len(filteredPod) > 0 {
+		if f.NextOnSuccess == nil && f.NextOnSuccessOrFailure == nil {
+			// No succeeding filters to run, return.
+			return filteredPod
+		}
+		if f.NextOnSuccess != nil {
+			next = f.NextOnSuccess
+		}
+		loggerTrace.Info("Filter succeeded", "filter", f.Name(), "next", next.Name(), "filteredPodCount", len(filteredPod))
+		// On success, pass the filtered result to the next filter.
+		return next.Filter(ctx, filteredPod)
+	} else {
+		if f.NextOnFailure == nil && f.NextOnSuccessOrFailure == nil {
+			// No succeeding filters to run, return.
+			return filteredPod
+		}
+		if f.NextOnFailure != nil {
+			next = f.NextOnFailure
+		}
+		loggerTrace.Info("Filter failed", "filter", f.Name(), "next", next.Name())
+		// On failure, pass the initial set of pods to the next filter.
+		return next.Filter(ctx, pods)
+	}
+}
diff --git a/pkg/epp/scheduling/plugins/filter/filter.go b/pkg/epp/scheduling/plugins/filter/filter.go
deleted file mode 100644
index 86620aa..0000000
--- a/pkg/epp/scheduling/plugins/filter/filter.go
+++ /dev/null
@@ -1,278 +0,0 @@
-/*
-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 filter
-
-import (
-	"math"
-	"math/rand"
-	"time"
-
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/config"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
-	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
-)
-
-type baseFilter struct {
-	name   string
-	filter filterFunc
-}
-
-func (f *baseFilter) Name() string {
-	if f == nil {
-		return "nil"
-	}
-	return f.name
-}
-
-func (f *baseFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-	loggerTrace := ctx.Logger.V(logutil.TRACE)
-	loggerTrace.Info("Running a filter", "name", f.Name(), "podCount", len(pods))
-
-	return f.filter(ctx, pods)
-}
-
-// DecisionTreeFilter applies current filterFunc, and then recursively applies next filters
-// depending success or failure of the current filter.
-// It can be used to construct a flow chart algorithm.
-type DecisionTreeFilter struct {
-	Current plugins.Filter
-	// NextOnSuccess filter will be applied after successfully applying the current filter.
-	// The filtered results will be passed to the next filter.
-	NextOnSuccess plugins.Filter
-	// NextOnFailure filter will be applied if current filter fails.
-	// The original input will be passed to the next filter.
-	NextOnFailure plugins.Filter
-	// NextOnSuccessOrFailure is a convenience field to configure the next filter regardless of the
-	// success or failure of the current filter.
-	// NOTE: When using NextOnSuccessOrFailure, both nextOnSuccess and nextOnFailure SHOULD be nil.
-	// However if that's not the case, nextOnSuccess and nextOnFailure will be used, instead of
-	// NextOnSuccessOrFailure,  in the success and failure scenarios, respectively.
-	NextOnSuccessOrFailure plugins.Filter
-}
-
-func (f *DecisionTreeFilter) Name() string {
-	if f == nil {
-		return "nil"
-	}
-	return f.Current.Name()
-}
-
-func (f *DecisionTreeFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-	loggerTrace := ctx.Logger.V(logutil.TRACE)
-	filtered := f.Current.Filter(ctx, pods)
-
-	next := f.NextOnSuccessOrFailure
-	if len(filtered) > 0 {
-		if f.NextOnSuccess == nil && f.NextOnSuccessOrFailure == nil {
-			// No succeeding filters to run, return.
-			return filtered
-		}
-		if f.NextOnSuccess != nil {
-			next = f.NextOnSuccess
-		}
-		loggerTrace.Info("Filter succeeded", "filter", f.Name(), "next", next.Name(), "filteredPodCount", len(filtered))
-		// On success, pass the filtered result to the next filter.
-		return next.Filter(ctx, filtered)
-	} else {
-		if f.NextOnFailure == nil && f.NextOnSuccessOrFailure == nil {
-			// No succeeding filters to run, return.
-			return filtered
-		}
-		if f.NextOnFailure != nil {
-			next = f.NextOnFailure
-		}
-		loggerTrace.Info("Filter failed", "filter", f.Name(), "next", next.Name())
-		// On failure, pass the initial set of pods to the next filter.
-		return next.Filter(ctx, pods)
-	}
-}
-
-// filterFunc filters a set of input pods to a subset.
-type filterFunc func(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod
-
-// toFilterFunc is a helper function to convert a per pod filter func to the FilterFunc.
-func toFilterFunc(pp podPredicate) filterFunc {
-	return func(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-		filtered := []types.Pod{}
-		for _, pod := range pods {
-			pass := pp(ctx.Req, pod)
-			if pass {
-				filtered = append(filtered, pod)
-			}
-		}
-
-		return filtered
-	}
-}
-
-var LeastQueueFilter = &baseFilter{
-	name:   "least queuing",
-	filter: leastQueuingFilterFunc,
-}
-
-// leastQueuingFilterFunc finds the max and min queue size of all pods, divides the whole range
-// (max-min) by the number of pods, and finds the pods that fall into the first range.
-// The intuition is that if there are multiple pods that share similar queue size in the low range,
-// we should consider them all instead of the absolute minimum one. This worked better than picking
-// the least one as it gives more choices for the next filter, which on aggregate gave better
-// results.
-// TODO: Compare this strategy with other strategies such as top K.
-func leastQueuingFilterFunc(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-	min := math.MaxInt
-	max := 0
-	filtered := []types.Pod{}
-
-	for _, pod := range pods {
-		if pod.GetMetrics().WaitingQueueSize <= min {
-			min = pod.GetMetrics().WaitingQueueSize
-		}
-		if pod.GetMetrics().WaitingQueueSize >= max {
-			max = pod.GetMetrics().WaitingQueueSize
-		}
-	}
-
-	for _, pod := range pods {
-		if pod.GetMetrics().WaitingQueueSize >= min && pod.GetMetrics().WaitingQueueSize <= min+(max-min)/len(pods) {
-			filtered = append(filtered, pod)
-		}
-	}
-	return filtered
-}
-
-var LowQueueFilter = &baseFilter{
-	name:   "low queueing filter",
-	filter: toFilterFunc((queueThresholdPredicate(config.Conf.QueueingThresholdLoRA))),
-}
-
-var LeastKVCacheFilter = &baseFilter{
-	name:   "least KV cache percent",
-	filter: leastKVCacheFilterFunc,
-}
-
-// leastKVCacheFilterFunc finds the max and min KV cache of all pods, divides the whole range
-// (max-min) by the number of pods, and finds the pods that fall into the first range.
-// The intuition is that if there are multiple pods that share similar KV cache in the low range, we
-// should consider them all instead of the absolute minimum one. This worked better than picking the
-// least one as it gives more choices for the next filter, which on aggregate gave better results.
-// TODO: Compare this strategy with other strategies such as top K.
-func leastKVCacheFilterFunc(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-	min := math.MaxFloat64
-	var max float64 = 0
-	filtered := []types.Pod{}
-
-	for _, pod := range pods {
-		if pod.GetMetrics().KVCacheUsagePercent <= min {
-			min = pod.GetMetrics().KVCacheUsagePercent
-		}
-		if pod.GetMetrics().KVCacheUsagePercent >= max {
-			max = pod.GetMetrics().KVCacheUsagePercent
-		}
-	}
-
-	for _, pod := range pods {
-		if pod.GetMetrics().KVCacheUsagePercent >= min && pod.GetMetrics().KVCacheUsagePercent <= min+(max-min)/float64(len(pods)) {
-			filtered = append(filtered, pod)
-		}
-	}
-	return filtered
-}
-
-var LoRAAffinityFilter = &baseFilter{
-	name:   "affinity LoRA",
-	filter: loRASoftAffinityFilterFunc,
-}
-
-// loRASoftAffinityPredicate implements a pod selection strategy that prioritizes pods
-// with existing LoRA model affinity while allowing for load balancing through randomization.
-//
-// The function works by:
-// 1. Separating pods into two groups: those with target model affinity and those with available capacity
-// 2. Using a probability threshold to sometimes select from non-affinity pods to enable load balancing
-// 3. Falling back to whatever group has pods if one group is empty
-//
-// Parameters:
-//   - logger: Logger interface for diagnostic output
-//   - req: LLM request containing the resolved target model
-//   - pods: Slice of pod metrics to filter
-//
-// Returns:
-//   - Filtered slice of pod metrics based on affinity and availability
-//   - Error if any issues occur during filtering
-func loRASoftAffinityFilterFunc(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-
-	// Pre-allocate slices with estimated capacity
-	filtered_affinity := make([]types.Pod, 0, len(pods))
-	filtered_available := make([]types.Pod, 0, len(pods))
-
-	// Categorize pods based on affinity and availability
-	for _, pod := range pods {
-		_, active := pod.GetMetrics().ActiveModels[ctx.Req.ResolvedTargetModel]
-		_, waiting := pod.GetMetrics().WaitingModels[ctx.Req.ResolvedTargetModel]
-
-		if active || waiting {
-			filtered_affinity = append(filtered_affinity, pod)
-		} else if len(pod.GetMetrics().ActiveModels)+len(pod.GetMetrics().WaitingModels) < pod.GetMetrics().MaxActiveModels {
-			filtered_available = append(filtered_available, pod)
-		}
-	}
-
-	// Use crypto/rand for better randomization in production environments
-	randSource := rand.NewSource(time.Now().UnixNano())
-	randGen := rand.New(randSource)
-
-	// If both groups have pods, use probability to select which group to return
-	if len(filtered_affinity) > 0 && len(filtered_available) > 0 {
-		if randGen.Float64() < config.Conf.LoraAffinityThreshold {
-			return filtered_affinity
-		}
-		return filtered_available
-	}
-
-	// Return whichever group has pods
-	if len(filtered_affinity) > 0 {
-		return filtered_affinity
-	}
-
-	return filtered_available
-}
-
-var HasCapacityFilter = &baseFilter{
-	name:   "has capacity for sheddable requests",
-	filter: toFilterFunc(queueThresholdPredicate(config.Conf.QueueThresholdCritical).and(kvCacheThresholdPredicate(config.Conf.KVCacheThreshold))),
-}
-
-// podPredicate is a filter function to check whether a pod is desired.
-type podPredicate func(req *types.LLMRequest, pod types.Pod) bool
-
-func queueThresholdPredicate(queueThreshold int) podPredicate {
-	return func(req *types.LLMRequest, pod types.Pod) bool {
-		return pod.GetMetrics().WaitingQueueSize <= queueThreshold
-	}
-}
-
-func kvCacheThresholdPredicate(kvCacheThreshold float64) podPredicate {
-	return func(req *types.LLMRequest, pod types.Pod) bool {
-		return pod.GetMetrics().KVCacheUsagePercent <= kvCacheThreshold
-	}
-}
-
-func (pp podPredicate) and(another podPredicate) podPredicate {
-	return func(req *types.LLMRequest, pod types.Pod) bool {
-		return pp(req, pod) && another(req, pod)
-	}
-}
diff --git a/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go b/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go
new file mode 100644
index 0000000..eed6476
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go
@@ -0,0 +1,68 @@
+/*
+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 filter
+
+import (
+	"math"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+// compile-time type validation
+var _ plugins.Filter = &LeastKVCacheFilter{}
+
+// NewLeastKVCacheFilter returns a new LeastKVCacheFilter.
+func NewLeastKVCacheFilter() *LeastKVCacheFilter {
+	return &LeastKVCacheFilter{}
+}
+
+// LeastKVCacheFilter finds the max and min KV cache of all pods, divides the whole range
+// (max-min) by the number of pods, and finds the pods that fall into the first range.
+// The intuition is that if there are multiple pods that share similar KV cache in the low range, we
+// should consider them all instead of the absolute minimum one. This worked better than picking the
+// least one as it gives more choices for the next filter, which on aggregate gave better results.
+type LeastKVCacheFilter struct{}
+
+// Name returns the name of the filter.
+func (f *LeastKVCacheFilter) Name() string {
+	return "least-KV-cache"
+}
+
+// Filter filters out pods that doesn't meet the filter criteria.
+func (f *LeastKVCacheFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	filteredPods := []types.Pod{}
+
+	min := math.MaxFloat64
+	var max float64 = 0
+
+	for _, pod := range pods {
+		if pod.GetMetrics().KVCacheUsagePercent <= min {
+			min = pod.GetMetrics().KVCacheUsagePercent
+		}
+		if pod.GetMetrics().KVCacheUsagePercent >= max {
+			max = pod.GetMetrics().KVCacheUsagePercent
+		}
+	}
+
+	for _, pod := range pods {
+		if pod.GetMetrics().KVCacheUsagePercent >= min && pod.GetMetrics().KVCacheUsagePercent <= min+(max-min)/float64(len(pods)) {
+			filteredPods = append(filteredPods, pod)
+		}
+	}
+	return filteredPods
+}
diff --git a/pkg/epp/scheduling/plugins/filter/least_queue_filter.go b/pkg/epp/scheduling/plugins/filter/least_queue_filter.go
new file mode 100644
index 0000000..7a71d51
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/filter/least_queue_filter.go
@@ -0,0 +1,70 @@
+/*
+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 filter
+
+import (
+	"math"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+// compile-time type validation
+var _ plugins.Filter = &LeastQueueFilter{}
+
+// NewLeastQueueFilter returns a new LeastQueueFilter.
+func NewLeastQueueFilter() *LeastQueueFilter {
+	return &LeastQueueFilter{}
+}
+
+// LeastQueueFilter finds the max and min queue size of all pods, divides the whole range
+// (max-min) by the number of pods, and finds the pods that fall into the first range.
+// The intuition is that if there are multiple pods that share similar queue size in the low range,
+// we should consider them all instead of the absolute minimum one. This worked better than picking
+// the least one as it gives more choices for the next filter, which on aggregate gave better
+// results.
+type LeastQueueFilter struct{}
+
+// Name returns the name of the filter.
+func (f *LeastQueueFilter) Name() string {
+	return "least-queue"
+}
+
+// Filter filters out pods that doesn't meet the filter criteria.
+func (f *LeastQueueFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	filteredPods := []types.Pod{}
+
+	min := math.MaxInt
+	max := 0
+
+	for _, pod := range pods {
+		if pod.GetMetrics().WaitingQueueSize <= min {
+			min = pod.GetMetrics().WaitingQueueSize
+		}
+		if pod.GetMetrics().WaitingQueueSize >= max {
+			max = pod.GetMetrics().WaitingQueueSize
+		}
+	}
+
+	for _, pod := range pods {
+		if pod.GetMetrics().WaitingQueueSize >= min && pod.GetMetrics().WaitingQueueSize <= min+(max-min)/len(pods) {
+			filteredPods = append(filteredPods, pod)
+		}
+	}
+
+	return filteredPods
+}
diff --git a/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go b/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go
new file mode 100644
index 0000000..bc744a8
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go
@@ -0,0 +1,90 @@
+/*
+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 filter
+
+import (
+	"math/rand"
+	"time"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/config"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+// compile-time type validation
+var _ plugins.Filter = &LoraAffinityFilter{}
+
+// NewLoraAffinityFilter returns a new LoraAffinityFilter.
+func NewLoraAffinityFilter() *LoraAffinityFilter {
+	return &LoraAffinityFilter{
+		loraAffinityThreshold: config.Conf.LoraAffinityThreshold,
+	}
+}
+
+// LoraAffinityFilter implements a pod selection strategy that prioritizes pods
+// with existing LoRA model affinity while allowing for load balancing through randomization.
+//
+// The function works by:
+// 1. Separating pods into two groups: those with target model affinity and those with available capacity
+// 2. Using a probability threshold to sometimes select from non-affinity pods to enable load balancing
+// 3. Falling back to whatever group has pods if one group is empty
+type LoraAffinityFilter struct {
+	loraAffinityThreshold float64
+}
+
+// Name returns the name of the filter.
+func (f *LoraAffinityFilter) Name() string {
+	return "lora-affinity"
+}
+
+// Filter filters out pods that doesn't meet the filter criteria.
+func (f *LoraAffinityFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	// Pre-allocate slices with estimated capacity
+	filtered_affinity := make([]types.Pod, 0, len(pods))
+	filtered_available := make([]types.Pod, 0, len(pods))
+
+	// Categorize pods based on affinity and availability
+	for _, pod := range pods {
+		_, active := pod.GetMetrics().ActiveModels[ctx.Req.TargetModel]
+		_, waiting := pod.GetMetrics().WaitingModels[ctx.Req.TargetModel]
+
+		if active || waiting {
+			filtered_affinity = append(filtered_affinity, pod)
+		} else if len(pod.GetMetrics().ActiveModels)+len(pod.GetMetrics().WaitingModels) < pod.GetMetrics().MaxActiveModels {
+			filtered_available = append(filtered_available, pod)
+		}
+	}
+
+	// Use crypto/rand for better randomization in production environments
+	randSource := rand.NewSource(time.Now().UnixNano())
+	randGen := rand.New(randSource)
+
+	// If both groups have pods, use probability to select which group to return
+	if len(filtered_affinity) > 0 && len(filtered_available) > 0 {
+		if randGen.Float64() < f.loraAffinityThreshold {
+			return filtered_affinity
+		}
+		return filtered_available
+	}
+
+	// Return whichever group has pods
+	if len(filtered_affinity) > 0 {
+		return filtered_affinity
+	}
+
+	return filtered_available
+}
diff --git a/pkg/epp/scheduling/plugins/filter/low_queue_filter.go b/pkg/epp/scheduling/plugins/filter/low_queue_filter.go
new file mode 100644
index 0000000..feb599b
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/filter/low_queue_filter.go
@@ -0,0 +1,56 @@
+/*
+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 filter
+
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/config"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+// compile-time type validation
+var _ plugins.Filter = &LowQueueFilter{}
+
+// NewLowQueueFilter returns a new LowQueueFilter.
+func NewLowQueueFilter() *LowQueueFilter {
+	return &LowQueueFilter{
+		queueingThresholdLoRA: config.Conf.QueueingThresholdLoRA,
+	}
+}
+
+// LowQueueFilter returns pods that their waiting queue size is less than a configured threshold
+type LowQueueFilter struct {
+	queueingThresholdLoRA int
+}
+
+// Name returns the name of the filter.
+func (f *LowQueueFilter) Name() string {
+	return "low-queue"
+}
+
+// Filter filters out pods that doesn't meet the filter criteria.
+func (f *LowQueueFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	filteredPods := []types.Pod{}
+
+	for _, pod := range pods {
+		if pod.GetMetrics().WaitingQueueSize <= f.queueingThresholdLoRA {
+			filteredPods = append(filteredPods, pod)
+		}
+	}
+
+	return filteredPods
+}
diff --git a/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go b/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go
new file mode 100644
index 0000000..5a298a0
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.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 filter
+
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/config"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+// compile-time type validation
+var _ plugins.Filter = &SheddableCapacityFilter{}
+
+// NewSheddableCapacityFilter returns a new SheddableCapacityFilter.
+func NewSheddableCapacityFilter() *SheddableCapacityFilter {
+	return &SheddableCapacityFilter{
+		queueThreshold:   config.Conf.QueueThresholdCritical,
+		kvCacheThreshold: config.Conf.KVCacheThreshold,
+	}
+}
+
+// SheddableCapacityFilter filters only pods that has capacity for sheddable requests.
+type SheddableCapacityFilter struct {
+	queueThreshold   int
+	kvCacheThreshold float64
+}
+
+// Name returns the name of the filter.
+func (f *SheddableCapacityFilter) Name() string {
+	return "sheddable-capacity"
+}
+
+// Filter filters out pods that doesn't meet the filter criteria.
+func (f *SheddableCapacityFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	if ctx.Req.Critical {
+		return pods // // Allow all pods to passthrough if the request is critical, even if all pods reach their capacity.
+	}
+
+	filteredPods := []types.Pod{}
+
+	for _, pod := range pods {
+		if pod.GetMetrics().WaitingQueueSize <= f.queueThreshold && pod.GetMetrics().KVCacheUsagePercent <= f.kvCacheThreshold {
+			filteredPods = append(filteredPods, pod)
+		}
+	}
+
+	return filteredPods
+}
diff --git a/pkg/epp/scheduling/plugins/prefix/indexer.go b/pkg/epp/scheduling/plugins/prefix/indexer.go
new file mode 100644
index 0000000..2017ba1
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/prefix/indexer.go
@@ -0,0 +1,173 @@
+/*
+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 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 prefix
+
+import (
+	"context"
+	"sync"
+	"time"
+	"unsafe"
+
+	"container/list"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+func newIndexer(maxCacheSize int) *indexer {
+	t := &indexer{
+		maxCacheSize: maxCacheSize,
+		table:        make(map[BlockHash]map[ServerID]*list.Element),
+		ll:           list.New(),
+	}
+	go t.ReportCacheSize(time.Second)
+	return t
+}
+
+// An indexer maintains an LRU cache of prompt prefix hashes and the server(s) that might have that
+// prefix cached .
+type indexer struct {
+	mu           sync.RWMutex
+	maxCacheSize int
+	table        map[BlockHash]map[ServerID]*list.Element // from any prefix cache to the cache entry to find the server
+	ll           *list.List                               // LinkedList to keep track of the order of entries
+}
+
+// value is the value stored in the linked list.
+type value struct {
+	server ServerID
+	hash   BlockHash
+}
+
+// Get returns the set of servers that have the given prefix hash cached.
+func (i *indexer) Get(hash BlockHash) map[ServerID]bool {
+	i.mu.RLock()
+	defer i.mu.RUnlock()
+	res := map[ServerID]bool{}
+	for server := range i.table[hash] {
+		res[server] = true
+	}
+	return res
+}
+
+// Add adds a list of prefix hashes of a single request to the server the request was sent to.
+// The intuition is that this server is likely to have the prefix cached, so next time a request
+// sharing the longest prefix should be sent to the same server to take advantage of the cache hit.
+func (i *indexer) Add(hashes []BlockHash, server ServerID) {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+	for _, hash := range hashes {
+		i.add(hash, server)
+	}
+}
+
+func (i *indexer) check(hash BlockHash, server ServerID) (*list.Element, bool) {
+	servers, ok := i.table[hash]
+	if !ok {
+		return nil, false
+	}
+	e, ok := servers[server]
+	return e, ok
+}
+
+func (i *indexer) add(hash BlockHash, server ServerID) {
+	e, exists := i.check(hash, server)
+	if exists {
+		i.ll.MoveToBack(e)
+	} else {
+		i.create(hash, server)
+	}
+}
+
+func (i *indexer) create(hash BlockHash, server ServerID) {
+	for i.ll.Len() >= i.maxCacheSize {
+		// Evict the least recently used entry if we've exceeded the max cache size
+		i.evict()
+	}
+
+	if _, ok := i.table[hash]; !ok {
+		i.table[hash] = make(map[ServerID]*list.Element)
+	}
+	v := &value{
+		server: server,
+		hash:   hash,
+	}
+	e := i.ll.PushBack(v)
+	i.table[hash][server] = e
+}
+
+// evict removes the least recently used entry from the cache
+func (i *indexer) evict() {
+	oldestNode := i.ll.Front()
+	if oldestNode == nil {
+		return
+	}
+	i.ll.Remove(oldestNode)
+
+	v := oldestNode.Value.(*value)
+	hash := v.hash
+	server := v.server
+	// Remove from the hash map
+	serverMap := i.table[hash]
+	delete(serverMap, server)
+
+	// If this was the last server for this hash, remove the hash entry entirely
+	if len(serverMap) == 0 {
+		delete(i.table, hash)
+	}
+
+	log.FromContext(context.TODO()).V(logutil.TRACE).Info("Evicted LRU entry", "hash", hash, "server", server)
+}
+
+// ReportCacheSize starts a goroutine that periodically reports the cache size metric
+func (i *indexer) ReportCacheSize(interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for range ticker.C {
+		i.mu.RLock()
+		metrics.RecordPrefixCacheSize(int64(i.ll.Len()))
+		log.FromContext(context.TODO()).V(logutil.TRACE).Info("LRU", "# entries", i.ll.Len(), "estimated size MB", i.ll.Len()*i.estimateEntrySize()/1000000)
+		i.mu.RUnlock()
+	}
+}
+
+// estimateEntrySize estimates the memory size of a cache entry in bytes.
+func (i *indexer) estimateEntrySize() int {
+	size := 0
+
+	// Estimate the size of a node in the linked list.
+	// First get the size of the node struct via unsafe.Sizeof.
+	// The prev and next pointers are 8 bytes each on a 64-bit system.
+	// The BlockHash is a uint64, which is 8 bytes.
+	// The ServerID is a NamespacedName, which contains two strings (Name and Namespace).
+	// The headers for the strings are 16 bytes each (8 bytes for the pointer and 8 bytes for the length).
+	// So unsafe.Sizeof(node{}) should return 2*8 + 8 + 2*16 = 48 bytes.
+	size += int(unsafe.Sizeof(value{}))
+	// Size of the Name and Namespace strings in ServerID, assuming 63 bytes each (max length for Kubernetes NamespacedName).
+	size += 2 * 63
+
+	// Estimate the size of an entry in the hash map. Note the overhead of the map headers and buckets are ignored.
+	size += 8      // Size of the BlockHash (uint64).
+	size += 2 * 16 // Size of the ServerID string headers (NamespacedName).
+	size += 2 * 63 // Size of the Name and Namespace strings in ServerID.
+	size += 8      // Size of the pointer to the node in the hash map.
+
+	// Based on the above estimates, the estimated size of an entry is:
+	// (48 + 2*63) + (8 + 2*16 + 2*63 + 8) = 348 bytes.
+	return size
+}
diff --git a/pkg/epp/scheduling/plugins/prefix/plugin.go b/pkg/epp/scheduling/plugins/prefix/plugin.go
new file mode 100644
index 0000000..5cb2d4c
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/prefix/plugin.go
@@ -0,0 +1,246 @@
+/*
+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 prefix
+
+import (
+	"encoding/binary"
+	"fmt"
+
+	"github.com/cespare/xxhash/v2"
+	k8stypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+const (
+	DefaultScorerWeight = 1
+	// Attempt to return DefaultNumServersToMatch servers with their longest prefix match length.
+	// Why not just return the server with longest prefix match?
+	// It may not be the optimal choice, e.g., it may have a high queue depth.
+	// We optimistically search more than one to give more candidates for the scheduler to choose.
+	DefaultNumServersToMatch = 2
+	// vLLM default token block size is 16, and a good guess of average characters per token is 4.
+	DefaultHashBlockSize = 64
+	// The maximum number of blocks to match. Two long requests with the same prefix up to this
+	// limit will be indistinguishable.
+	// This parameter provides a trade-off between cache size, prefix matching speed and matching
+	// accuracy. Use a small value if most requests are short to reduce cache size and speed up the
+	// matching process. Use a large value if most requests are long to increase the matching accuracy.
+	DefaultMaxPrefixBlocks = 128
+	// The indexer is an approximation to the actual prefix cache state on the model servers.
+	// A small capacity ensures a high accuracy of cache hit on the model server, but it will
+	// increase the chance of false negatives. A high capacity does the opposite.
+	// To properly size this, consider the sum of the total number of cache entries on all model
+	// servers. Consider the llama3 8B model on 3 H100 80GB GPUs. The size of the model weight is
+	// about 16GB. Assume 50% of the remaining HBM is used for caching prefixes, we have 32GB. Each
+	// token is about 128KB in size, so we can cache 250K tokens. Using the default block size of 16
+	// in vLLM, we will have 250K / 16 = 15.6K blocks. In total we have 15.6K * 3 = 46.8K blocks, or
+	// roughly 50K.
+	// How much memory space does it require to hold the 50K block hashes?
+	// According to the estimates in indexer.estimateEntrySize(), the size of each entry is
+	// approximately 348 bytes. So in total we have 50K * 348 = 17.4MB.
+	DefaultLRUIndexerCapacity = 50000
+)
+
+type Config struct {
+	// The input prompt is broken into sizes of HashBlockSize to calculate block hashes . Requests
+	// with length shorter than the block size will be ignored.
+	HashBlockSize int
+	// MaxPrefixBlocksToMatch is the maximum number of prefix blocks to match. Input beyond this limit will
+	// be ignored.
+	MaxPrefixBlocksToMatch int
+	// Max (approximate) size of the LRU indexer in number of entries.
+	LRUIndexerCapacity int
+}
+
+type Plugin struct {
+	Config
+	indexer Indexer
+}
+
+type Indexer interface {
+	Get(hash BlockHash) map[ServerID]bool
+	Add(hashes []BlockHash, server ServerID)
+}
+
+// BlockHash is a hash of the block of request body.
+type BlockHash uint64
+
+type ServerID k8stypes.NamespacedName
+
+func (s ServerID) String() string {
+	return k8stypes.NamespacedName(s).String()
+}
+
+var _ types.StateData = &schedulingContextState{}
+
+// This is the state of this plugin to be used during a scheduling cycle.
+type schedulingContextState struct {
+	// PrefixHashes is a list of prefix hashes of the request prompt broken into blocks.
+	PrefixHashes []BlockHash
+	// A map of server to its longest prefix cache match length.
+	PrefixCacheServers map[ServerID]int
+}
+
+func (s *schedulingContextState) Clone() types.StateData {
+	prefixHashes := make([]BlockHash, len(s.PrefixHashes))
+	copy(prefixHashes, s.PrefixHashes)
+	prefixCacheServers := make(map[ServerID]int, len(s.PrefixCacheServers))
+	for key, value := range s.PrefixCacheServers {
+		prefixCacheServers[key] = value
+	}
+
+	return &schedulingContextState{
+		PrefixHashes:       prefixHashes,
+		PrefixCacheServers: prefixCacheServers,
+	}
+}
+
+func New(config Config) *Plugin {
+	m := &Plugin{
+		Config:  config,
+		indexer: newIndexer(config.LRUIndexerCapacity),
+	}
+	return m
+}
+
+func (m *Plugin) Name() string {
+	return "prefix-cache"
+}
+
+func (m *Plugin) PreSchedule(ctx *types.SchedulingContext) {
+	hashes := hashPrompt(ctx, m.HashBlockSize, m.MaxPrefixBlocksToMatch)
+	state := &schedulingContextState{
+		PrefixHashes:       hashes,
+		PrefixCacheServers: m.matchLongestPrefix(ctx, hashes, DefaultNumServersToMatch),
+	}
+
+	ctx.CycleState.Write(types.StateKey(m.Name()), state)
+	ctx.Logger.V(logutil.DEBUG).Info(fmt.Sprintf("PreSchedule, cached servers: %+v", state.PrefixCacheServers), "hashes", state.PrefixHashes)
+}
+
+// If a request was routed to a server, record it in the cache:
+func (m *Plugin) PostSchedule(ctx *types.SchedulingContext, res *types.Result) {
+	targetPod := res.TargetPod.GetPod()
+	state, err := m.getPrefixState(ctx.CycleState)
+	if err != nil {
+		ctx.Logger.Error(err, "failed to read prefix plugin cycle state")
+		return
+	}
+	m.indexer.Add(state.PrefixHashes, ServerID(targetPod.NamespacedName))
+	total := len(state.PrefixHashes)
+	matchLen := state.PrefixCacheServers[ServerID(targetPod.NamespacedName)]
+	metrics.RecordPrefixCacheMatch(matchLen*m.HashBlockSize, total*m.HashBlockSize)
+}
+
+func (m *Plugin) Score(ctx *types.SchedulingContext, pods []types.Pod) map[types.Pod]float64 {
+	scores := make(map[types.Pod]float64, len(pods))
+
+	state, err := m.getPrefixState(ctx.CycleState)
+	if err != nil {
+		ctx.Logger.Error(err, "failed to read prefix plugin cycle state")
+		return scores
+	}
+
+	total := len(state.PrefixHashes)
+	podScoreFunc := func(pod types.Pod) float64 {
+		if total == 0 {
+			return 0
+		}
+		matchLen := state.PrefixCacheServers[ServerID(pod.GetPod().NamespacedName)]
+		return float64(matchLen) / float64(total)
+	}
+
+	for _, pod := range pods {
+		scores[pod] = podScoreFunc(pod)
+	}
+	return scores
+}
+
+// matchLongestPrefix returns a map of servers and length of prefix that each server caches.
+func (m *Plugin) matchLongestPrefix(ctx *types.SchedulingContext, hashes []BlockHash, numServers int) map[ServerID]int {
+	if numServers > len(ctx.PodsSnapshot) {
+		numServers = len(ctx.PodsSnapshot)
+	}
+	res := make(map[ServerID]int)
+	// Use a greedy strategy to search from the longest prefix.
+	// NOTE: It's possible to further optimize this with a binary search.
+	for i := len(hashes) - 1; i >= 0 && len(res) < numServers; i-- {
+		hash := hashes[i]
+		cachedServers := m.indexer.Get(hash)
+		if len(cachedServers) > 0 {
+			ctx.Logger.V(logutil.DEBUG).Info("Found cached servers", "cachedServers", cachedServers, "total # blocks", len(hashes), "longest prefix", i)
+			for server := range cachedServers {
+				// Update servers with their longest prefix match.
+				// If we already found this server with longer prefix match, don't update it.
+				if _, ok := res[server]; !ok {
+					res[server] = i + 1
+				}
+			}
+		}
+	}
+	return res
+}
+
+func (m *Plugin) getPrefixState(cycleState *types.CycleState) (*schedulingContextState, error) {
+	prefixStateKey := types.StateKey(m.Name())
+	state, err := cycleState.Read(prefixStateKey)
+	if err != nil {
+		return nil, fmt.Errorf("failed reading %q from CycleState: %w", prefixStateKey, err)
+	}
+
+	prefixSchedulingState, ok := state.(*schedulingContextState)
+	if !ok {
+		return nil, fmt.Errorf("invalid Prefix state, got type %T", state)
+	}
+
+	return prefixSchedulingState, nil
+}
+
+// hashPrompt divides the prompt into blocks and calculate the prefix cache for each block.
+// hash(0) is the hash of the model name, since different models generally don't share prefix cache.
+// For block i, hash(i) = hash(block i content, hash(i-1)).
+func hashPrompt(ctx *types.SchedulingContext, cacheBlockSize int, maxPrefixBlocks int) []BlockHash {
+	prompt := []byte(ctx.Req.Prompt)
+	if len(prompt) < cacheBlockSize {
+		ctx.Logger.V(logutil.DEBUG).Info("Request body too small for prefix cache", "size", len(prompt), "block size", cacheBlockSize)
+		return nil
+	}
+	if len(prompt) > cacheBlockSize*maxPrefixBlocks {
+		ctx.Logger.V(logutil.DEBUG).Info("Truncating input", "size", len(prompt), "max prefix blocks", maxPrefixBlocks, "block size", cacheBlockSize)
+		prompt = prompt[:maxPrefixBlocks*cacheBlockSize]
+	}
+	// Split the body into blocks of size cacheBlockSize. The +1 is to account for the model.
+	// If the last block is smaller than cacheBlockSize, it will be ignored.
+	res := make([]BlockHash, 0, 1+len(prompt)/cacheBlockSize)
+	// Add the model to the first block hash so that different models have different hashes even with the same body.
+	res = append(res, BlockHash(xxhash.Sum64String(ctx.Req.TargetModel)))
+	for i := 0; i+cacheBlockSize <= len(prompt); i += cacheBlockSize {
+		block := prompt[i : i+cacheBlockSize]
+		prevBlockHash := res[len(res)-1]
+		block = append(block, toBytes(prevBlockHash)...)
+		res = append(res, BlockHash(xxhash.Sum64(block)))
+	}
+	return res
+}
+
+func toBytes(i BlockHash) []byte {
+	bytes := make([]byte, 8)
+	binary.LittleEndian.PutUint64(bytes, uint64(i))
+	return bytes
+}
diff --git a/pkg/epp/scheduling/plugins/scorer/kvcache.go b/pkg/epp/scheduling/plugins/scorer/kvcache.go
index 0877691..dbb6079 100644
--- a/pkg/epp/scheduling/plugins/scorer/kvcache.go
+++ b/pkg/epp/scheduling/plugins/scorer/kvcache.go
@@ -20,6 +20,10 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
+const (
+	DefaultKVCacheScorerWeight = 1
+)
+
 type KVCacheScorer struct{}
 
 func (ss *KVCacheScorer) Name() string {
diff --git a/pkg/epp/scheduling/plugins/scorer/queue.go b/pkg/epp/scheduling/plugins/scorer/queue.go
index 3df9d41..bbe6b69 100644
--- a/pkg/epp/scheduling/plugins/scorer/queue.go
+++ b/pkg/epp/scheduling/plugins/scorer/queue.go
@@ -22,6 +22,10 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
+const (
+	DefaultQueueScorerWeight = 1
+)
+
 type QueueScorer struct{}
 
 func (q *QueueScorer) Name() string {
diff --git a/pkg/epp/scheduling/scheduler.go b/pkg/epp/scheduling/scheduler.go
index 9215489..29ffee8 100644
--- a/pkg/epp/scheduling/scheduler.go
+++ b/pkg/epp/scheduling/scheduler.go
@@ -33,44 +33,49 @@ import (
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
-var (
-	lowLatencyFilter = &filter.DecisionTreeFilter{
-		Current: filter.LowQueueFilter,
+// NewScheduler returns a new scheduler with default scheduler plugins configuration.
+func NewScheduler(datastore Datastore) *Scheduler {
+	// When the scheduler is initialized with NewScheduler function, thw below config will be used as default.
+	// it's possible to call NewSchedulerWithConfig to pass a different scheduler config.
+	// For build time plugins changes, it's recommended to call in main.go to NewSchedulerWithConfig.
+	loraAffinityFilter := filter.NewLoraAffinityFilter()
+	leastQueueFilter := filter.NewLeastQueueFilter()
+	leastKvCacheFilter := filter.NewLeastKVCacheFilter()
+
+	lowLatencyFilter := &filter.DecisionTreeFilter{
+		Current: filter.NewLowQueueFilter(),
 		NextOnSuccess: &filter.DecisionTreeFilter{
-			Current: filter.LoRAAffinityFilter,
+			Current: loraAffinityFilter,
 			NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-				Current: filter.LeastQueueFilter,
+				Current: leastQueueFilter,
 				NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-					Current: filter.LeastKVCacheFilter,
+					Current: leastKvCacheFilter,
 				},
 			},
 		},
 		NextOnFailure: &filter.DecisionTreeFilter{
-			Current: filter.LeastQueueFilter,
+			Current: leastQueueFilter,
 			NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-				Current: filter.LoRAAffinityFilter,
+				Current: loraAffinityFilter,
 				NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-					Current: filter.LeastKVCacheFilter,
+					Current: leastKvCacheFilter,
 				},
 			},
 		},
 	}
 
-	sheddableRequestFilter = &filter.DecisionTreeFilter{
-		// When there is at least one model server that's not queuing requests, and still has KV
-		// cache below a certain threshold, we consider this model server has capacity to handle
-		// a sheddable request without impacting critical requests.
-		Current:       filter.HasCapacityFilter,
-		NextOnSuccess: lowLatencyFilter,
-		// If all pods are queuing or running above the KVCache threshold, we drop the sheddable
-		// request to make room for critical requests. for this, we don't define nextOnFailure.
+	defaultConfig := &SchedulerConfig{
+		preSchedulePlugins:  []plugins.PreSchedule{},
+		filters:             []plugins.Filter{filter.NewSheddableCapacityFilter(), lowLatencyFilter},
+		scorers:             map[plugins.Scorer]int{},
+		picker:              &picker.RandomPicker{},
+		postSchedulePlugins: []plugins.PostSchedule{},
 	}
-)
 
-func NewScheduler(datastore Datastore) *Scheduler {
 	return NewSchedulerWithConfig(datastore, defaultConfig)
 }
 
+// NewSchedulerWithConfig returns a new scheduler with the given scheduler plugins configuration.
 func NewSchedulerWithConfig(datastore Datastore, config *SchedulerConfig) *Scheduler {
 	return &Scheduler{
 		datastore:           datastore,
@@ -79,6 +84,7 @@ func NewSchedulerWithConfig(datastore Datastore, config *SchedulerConfig) *Sched
 		scorers:             config.scorers,
 		picker:              config.picker,
 		postSchedulePlugins: config.postSchedulePlugins,
+		postResponsePlugins: config.postResponsePlugins,
 	}
 }
 
@@ -89,6 +95,7 @@ type Scheduler struct {
 	scorers             map[plugins.Scorer]int // map from scorer to its weight
 	picker              plugins.Picker
 	postSchedulePlugins []plugins.PostSchedule
+	postResponsePlugins []plugins.PostResponse
 }
 
 type Datastore interface {
@@ -108,7 +115,7 @@ func (s *Scheduler) Schedule(ctx context.Context, req *types.LLMRequest) (*types
 	// Snapshot pod metrics from the datastore to:
 	// 1. Reduce concurrent access to the datastore.
 	// 2. Ensure consistent data during the scheduling operation of a request.
-	sCtx := types.NewSchedulingContext(ctx, req, types.ToSchedulerPodMetrics(s.datastore.PodGetAll()))
+	sCtx := types.NewSchedulingContext(ctx, req, nil, types.ToSchedulerPodMetrics(s.datastore.PodGetAll()))
 	loggerDebug.Info(fmt.Sprintf("Scheduling a request, Metrics: %+v", sCtx.PodsSnapshot))
 
 	s.runPreSchedulePlugins(sCtx)
@@ -207,18 +214,31 @@ func (s *Scheduler) runPostSchedulePlugins(ctx *types.SchedulingContext, res *ty
 	}
 }
 
-type defaultPlugin struct {
-	picker.RandomPicker
-}
+// OnResponse is invoked during the processing of a response from an inference pod. It will invoke
+// any defined plugins that process the response.
+func (s *Scheduler) OnResponse(ctx context.Context, resp *types.LLMResponse, targetPodName string) {
+	// Snapshot pod metrics from the datastore to:
+	// 1. Reduce concurrent access to the datastore.
+	// 2. Ensure consistent data during the scheduling operation of a request.
+	pods := types.ToSchedulerPodMetrics(s.datastore.PodGetAll())
+	var targetPod types.Pod
+	for _, pod := range pods {
+		if pod.GetPod().NamespacedName.String() == targetPodName {
+			targetPod = pod
+			break
+		}
+	}
 
-func (p *defaultPlugin) Name() string {
-	return "DefaultPlugin"
+	sCtx := types.NewSchedulingContext(ctx, nil, resp, pods)
+
+	s.runPostResponsePlugins(sCtx, targetPod)
 }
 
-func (p *defaultPlugin) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-	if ctx.Req.Critical {
-		return lowLatencyFilter.Filter(ctx, pods)
+func (s *Scheduler) runPostResponsePlugins(ctx *types.SchedulingContext, targetPod types.Pod) {
+	for _, plugin := range s.postResponsePlugins {
+		ctx.Logger.V(logutil.DEBUG).Info("Running post-response plugin", "plugin", plugin.Name())
+		before := time.Now()
+		plugin.PostResponse(ctx, targetPod)
+		metrics.RecordSchedulerPluginProcessingLatency(plugins.PostResponsePluginType, plugin.Name(), time.Since(before))
 	}
-
-	return sheddableRequestFilter.Filter(ctx, pods)
 }
diff --git a/pkg/epp/scheduling/types/cycle_state.go b/pkg/epp/scheduling/types/cycle_state.go
new file mode 100644
index 0000000..9f0a67f
--- /dev/null
+++ b/pkg/epp/scheduling/types/cycle_state.go
@@ -0,0 +1,92 @@
+/*
+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 types
+
+import (
+	"errors"
+	"sync"
+)
+
+var (
+	// ErrNotFound is the not found error message.
+	ErrNotFound = errors.New("not found")
+)
+
+// StateData is a generic type for arbitrary data stored in CycleState.
+type StateData interface {
+	// Clone is an interface to make a copy of StateData.
+	Clone() StateData
+}
+
+// StateKey is the type of keys stored in CycleState.
+type StateKey string
+
+// NewCycleState initializes a new CycleState and returns its pointer.
+func NewCycleState() *CycleState {
+	return &CycleState{}
+}
+
+// CycleState provides a mechanism for plugins to store and retrieve arbitrary data.
+// StateData stored by one plugin can be read, altered, or deleted by another plugin.
+// CycleState does not provide any data protection, as all plugins are assumed to be
+// trusted.
+// Note: CycleState uses a sync.Map to back the storage, because it is thread safe. It's aimed to optimize for the "write once and read many times" scenarios.
+type CycleState struct {
+	// key: StateKey, value: StateData
+	storage sync.Map
+}
+
+// Clone creates a copy of CycleState and returns its pointer. Clone returns
+// nil if the context being cloned is nil.
+func (c *CycleState) Clone() *CycleState {
+	if c == nil {
+		return nil
+	}
+	copy := NewCycleState()
+	// Safe copy storage in case of overwriting.
+	c.storage.Range(func(k, v interface{}) bool {
+		copy.storage.Store(k, v.(StateData).Clone())
+		return true
+	})
+
+	return copy
+}
+
+// Read retrieves data with the given "key" from CycleState. If the key is not
+// present, ErrNotFound is returned.
+//
+// See CycleState for notes on concurrency.
+func (c *CycleState) Read(key StateKey) (StateData, error) {
+	if v, ok := c.storage.Load(key); ok {
+		return v.(StateData), nil
+	}
+	return nil, ErrNotFound
+}
+
+// Write stores the given "val" in CycleState with the given "key".
+//
+// See CycleState for notes on concurrency.
+func (c *CycleState) Write(key StateKey, val StateData) {
+	c.storage.Store(key, val)
+}
+
+// Delete deletes data with the given key from CycleState.
+//
+// See CycleState for notes on concurrency.
+func (c *CycleState) Delete(key StateKey) {
+	c.storage.Delete(key)
+}
diff --git a/pkg/epp/scheduling/types/scheduling_context.go b/pkg/epp/scheduling/types/scheduling_context.go
new file mode 100644
index 0000000..3762180
--- /dev/null
+++ b/pkg/epp/scheduling/types/scheduling_context.go
@@ -0,0 +1,49 @@
+/*
+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 types
+
+import (
+	"context"
+
+	"github.com/go-logr/logr"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+func NewSchedulingContext(ctx context.Context, req *LLMRequest, resp *LLMResponse, pods []Pod) *SchedulingContext {
+
+	logger := log.FromContext(ctx).WithValues("request", req)
+	return &SchedulingContext{
+		Context:      ctx,
+		Logger:       logger,
+		Req:          req,
+		Resp:         resp,
+		PodsSnapshot: pods,
+		CycleState:   NewCycleState(),
+	}
+}
+
+// SchedulingContext holds contextual information during a scheduling operation.
+type SchedulingContext struct {
+	context.Context
+	Logger       logr.Logger
+	Req          *LLMRequest
+	Resp         *LLMResponse
+	PodsSnapshot []Pod
+	// CycleState can be used by plugins to store state during a scheduling cycle, to communicate
+	// between different extension points.
+	CycleState *CycleState
+}
diff --git a/pkg/epp/scheduling/types/types.go b/pkg/epp/scheduling/types/types.go
index 795ef65..a112794 100644
--- a/pkg/epp/scheduling/types/types.go
+++ b/pkg/epp/scheduling/types/types.go
@@ -17,21 +17,18 @@ limitations under the License.
 package types
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/go-logr/logr"
-	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 )
 
 // LLMRequest is a structured representation of the fields we parse out of the LLMRequest body.
 type LLMRequest struct {
-	// Model is the name of the model that the user specified in the request body.
-	Model string
-	// ResolvedTargetModel is the final target model after traffic split.
-	ResolvedTargetModel string
+	// TargetModel is the final target model after traffic split.
+	TargetModel string
+	// RequestId is the Envoy generated Id for the request being processed
+	RequestId string
 	// Critical is a boolean that specifies if a request is critical or not.
 	Critical bool
 	// Prompt is the prompt that was sent in the request body.
@@ -41,8 +38,21 @@ type LLMRequest struct {
 }
 
 func (r *LLMRequest) String() string {
-	return fmt.Sprintf("Model: %s, ResolvedTargetModel: %s, Critical: %t, PromptLength: %d, Headers: %v",
-		r.Model, r.ResolvedTargetModel, r.Critical, len(r.Prompt), r.Headers)
+	return fmt.Sprintf("TargetModel: %s, Critical: %t, PromptLength: %d, Headers: %v", r.TargetModel, r.Critical, len(r.Prompt), r.Headers)
+}
+
+// LLMResponse contains information from the response received to be passed to plugins
+type LLMResponse struct {
+	// RequestId is the Envoy generated Id for the request being processed
+	RequestId string
+	// Headers is a map of the response headers. Nil during body processing
+	Headers map[string]string
+	// Body Is the body of the response or nil during header processing
+	Body string
+	// IsStreaming indicates whether or not the response is being streamed by the model
+	IsStreaming bool
+	// EndOfStream when true indicates that this invocation contains the last chunk of the response
+	EndOfStream bool
 }
 
 type Pod interface {
@@ -56,14 +66,6 @@ type ScoredPod struct {
 	Score float64
 }
 
-// SchedulingContext holds contextual information during a scheduling operation.
-type SchedulingContext struct {
-	context.Context
-	Logger       logr.Logger
-	Req          *LLMRequest
-	PodsSnapshot []Pod
-}
-
 func (pm *PodMetrics) String() string {
 	if pm == nil {
 		return ""
@@ -84,16 +86,6 @@ type PodMetrics struct {
 	*backendmetrics.Metrics
 }
 
-func NewSchedulingContext(ctx context.Context, req *LLMRequest, pods []Pod) *SchedulingContext {
-	logger := log.FromContext(ctx).WithValues("request", req)
-	return &SchedulingContext{
-		Context:      ctx,
-		Logger:       logger,
-		Req:          req,
-		PodsSnapshot: pods,
-	}
-}
-
 func ToSchedulerPodMetrics(pods []backendmetrics.PodMetrics) []Pod {
 	pm := make([]Pod, 0, len(pods))
 	for _, pod := range pods {
diff --git a/pkg/epp/server/runserver.go b/pkg/epp/server/runserver.go
index 8abb3b1..4b86208 100644
--- a/pkg/epp/server/runserver.go
+++ b/pkg/epp/server/runserver.go
@@ -35,6 +35,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/controller"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/requestcontrol"
 )
 
 // ExtProcServerRunner provides methods to manage an external process server.
@@ -48,7 +49,7 @@ type ExtProcServerRunner struct {
 	CertPath                                 string
 	UseStreaming                             bool
 	RefreshPrometheusMetricsInterval         time.Duration
-	Scheduler                                handlers.Scheduler
+	Scheduler                                requestcontrol.Scheduler
 
 	// This should only be used in tests. We won't need this once we don't inject metrics in the tests.
 	// TODO:(https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/432) Cleanup
@@ -137,7 +138,7 @@ func (r *ExtProcServerRunner) AsRunnable(logger logr.Logger) manager.Runnable {
 		} else {
 			srv = grpc.NewServer()
 		}
-		extProcServer := handlers.NewStreamingServer(r.Scheduler, r.DestinationEndpointHintMetadataNamespace, r.DestinationEndpointHintKey, r.Datastore)
+		extProcServer := handlers.NewStreamingServer(r.DestinationEndpointHintMetadataNamespace, r.DestinationEndpointHintKey, r.Datastore, requestcontrol.NewDirector(r.Datastore, r.Scheduler))
 		extProcPb.RegisterExternalProcessorServer(
 			srv,
 			extProcServer,
diff --git a/pkg/epp/util/env/env.go b/pkg/epp/util/env/env.go
index 0c6d1c6..7ea7c3c 100644
--- a/pkg/epp/util/env/env.go
+++ b/pkg/epp/util/env/env.go
@@ -1,61 +1,53 @@
 package env
 
 import (
+	"fmt"
 	"os"
+	"reflect"
 	"strconv"
+	"time"
 
 	"github.com/go-logr/logr"
 )
 
-// getEnvFloat gets a float64 from an environment variable with a default value
-func GetEnvFloat(key string, defaultVal float64, logger logr.Logger) float64 {
-	val, exists := os.LookupEnv(key)
+// getEnvWithParser retrieves an environment variable. If set, it uses the provided parser to parse it.
+// It logs success or failure and returns the parsed value or the default value in case of a failure.
+func getEnvWithParser[T any](key string, defaultVal T, parser func(string) (T, error), logger logr.Logger) T {
+	valueStr, exists := os.LookupEnv(key)
 	if !exists {
-		logger.Info("Environment variable not set, using default value",
-			"key", key, "defaultValue", defaultVal)
+		logger.Info("Environment variable not set, using default value", "key", key, "defaultValue", defaultVal)
 		return defaultVal
 	}
 
-	floatVal, err := strconv.ParseFloat(val, 64)
+	parsedValue, err := parser(valueStr)
 	if err != nil {
-		logger.Info("Failed to parse environment variable as float, using default value",
-			"key", key, "value", val, "error", err, "defaultValue", defaultVal)
+		logger.Info(fmt.Sprintf("Failed to parse environment variable as %s, using default value", reflect.TypeOf(defaultVal)),
+			"key", key, "rawValue", valueStr, "error", err, "defaultValue", defaultVal)
 		return defaultVal
 	}
 
-	logger.Info("Successfully loaded environment variable",
-		"key", key, "value", floatVal)
-	return floatVal
+	logger.Info("Successfully loaded environment variable", "key", key, "value", parsedValue)
+	return parsedValue
 }
 
-// getEnvInt gets an int from an environment variable with a default value
-func GetEnvInt(key string, defaultVal int, logger logr.Logger) int {
-	val, exists := os.LookupEnv(key)
-	if !exists {
-		logger.Info("Environment variable not set, using default value",
-			"key", key, "defaultValue", defaultVal)
-		return defaultVal
-	}
+// GetEnvFloat gets a float64 from an environment variable with a default value.
+func GetEnvFloat(key string, defaultVal float64, logger logr.Logger) float64 {
+	parser := func(s string) (float64, error) { return strconv.ParseFloat(s, 64) }
+	return getEnvWithParser(key, defaultVal, parser, logger)
+}
 
-	intVal, err := strconv.Atoi(val)
-	if err != nil {
-		logger.Info("Failed to parse environment variable as int, using default value",
-			"key", key, "value", val, "error", err, "defaultValue", defaultVal)
-		return defaultVal
-	}
+// GetEnvInt gets an int from an environment variable with a default value.
+func GetEnvInt(key string, defaultVal int, logger logr.Logger) int {
+	return getEnvWithParser(key, defaultVal, strconv.Atoi, logger)
+}
 
-	logger.Info("Successfully loaded environment variable",
-		"key", key, "value", intVal)
-	return intVal
+// GetEnvDuration gets a time.Duration from an environment variable with a default value.
+func GetEnvDuration(key string, defaultVal time.Duration, logger logr.Logger) time.Duration {
+	return getEnvWithParser(key, defaultVal, time.ParseDuration, logger)
 }
 
-// GetEnvString gets a string from an environment variable with a default value
+// GetEnvString gets a string from an environment variable with a default value.
 func GetEnvString(key string, defaultVal string, logger logr.Logger) string {
-	val, exists := os.LookupEnv(key)
-	if !exists {
-		logger.Info("Environment variable not set, using default value",
-			"key", key, "defaultValue", defaultVal)
-		return defaultVal
-	}
-	return val
+	parser := func(s string) (string, error) { return s, nil }
+	return getEnvWithParser(key, defaultVal, parser, logger)
 }
diff --git a/site-src/guides/index.md b/site-src/guides/index.md
index be1b972..8981126 100644
--- a/site-src/guides/index.md
+++ b/site-src/guides/index.md
@@ -76,7 +76,7 @@ This quickstart guide is intended for engineers familiar with k8s and model serv
    kubectl apply -f https://github.com/kubernetes-sigs/gateway-api-inference-extension/raw/main/config/manifests/inferencemodel.yaml
    ```
 
-### Deploy the InferencePool and Extension
+### Deploy the InferencePool and Endpoint Picker Extension
 
    ```bash
    kubectl apply -f https://github.com/kubernetes-sigs/gateway-api-inference-extension/raw/main/config/manifests/inferencepool-resources.yaml
diff --git a/site-src/performance/benchmark/index.md b/site-src/performance/benchmark/index.md
index 39457bf..160cc26 100644
--- a/site-src/performance/benchmark/index.md
+++ b/site-src/performance/benchmark/index.md
@@ -1,45 +1,49 @@
 # Benchmark
 
-This user guide shows how to run benchmarks against a vLLM deployment, by using both the Gateway API
-inference extension, and a Kubernetes service as the load balancing strategy. The
-benchmark uses the [Latency Profile Generator](https://github.com/AI-Hypercomputer/inference-benchmark) (LPG)
-tool to generate load and collect results.
+This user guide shows how to run benchmarks against a vLLM model server deployment by using both Gateway API
+Inference Extension, and a Kubernetes service as the load balancing strategy. The benchmark uses the
+[Latency Profile Generator](https://github.com/AI-Hypercomputer/inference-benchmark) (LPG) tool to generate
+load and collect results.
 
 ## Prerequisites
 
 ### Deploy the inference extension and sample model server
 
-Follow this user guide https://gateway-api-inference-extension.sigs.k8s.io/guides/ to deploy the
-sample vLLM application, and the inference extension.
+Follow the [getting started guide](https://gateway-api-inference-extension.sigs.k8s.io/guides/#getting-started-with-gateway-api-inference-extension)
+to deploy the vLLM model server, CRDs, etc.
+
+__Note:__ Only the GPU-based model server deployment option is supported for benchmark testing.
 
 ### [Optional] Scale the sample vLLM deployment
 
-You will more likely to see the benefits of the inference extension when there are a decent number of replicas to make the optimal routing decision. 
+You are more likely to see the benefits of the inference extension when there are a decent number of replicas to make the optimal routing decision.
 
 ```bash
-kubectl scale --replicas=8 -f https://github.com/kubernetes-sigs/gateway-api-inference-extension/raw/main/config/manifests/vllm/gpu-deployment.yaml
+kubectl scale deployment vllm-llama3-8b-instruct --replicas=8
 ```
 
 ### Expose the model server via a k8s service
 
-As the baseline, let's also expose the vLLM deployment as a k8s service:
+To establish a baseline, expose the vLLM deployment as a k8s service:
 
 ```bash
-kubectl expose -f https://github.com/kubernetes-sigs/gateway-api-inference-extension/raw/main/config/manifests/vllm/gpu-deployment.yaml --port=8081 --target-port=8000 --type=LoadBalancer
+kubectl expose deployment vllm-llama3-8b-instruct --port=80 --target-port=8000 --type=LoadBalancer
 ```
 
 ## Run benchmark
 
-The LPG benchmark tool works by sending traffic to the specified target IP and port, and collect results. Follow the steps below to run a single benchmark. You can deploy multiple LPG instances if you want to run benchmarks in parallel against different targets.
+The LPG benchmark tool works by sending traffic to the specified target IP and port, and collecting the results.
+Follow the steps below to run a single benchmark. Multiple LPG instances can be deployed to run benchmarks in
+parallel against different targets.
 
 1. Check out the repo.
-    
+
     ```bash
     git clone https://github.com/kubernetes-sigs/gateway-api-inference-extension
     cd gateway-api-inference-extension
     ```
 
-1. Get the target IP. Examples below show how to get the IP of a gateway or a LoadBalancer k8s service.
+1. Get the target IP. The examples below shows how to get the IP of a gateway or a k8s service.
 
     ```bash
     # Get gateway IP
@@ -51,32 +55,43 @@ The LPG benchmark tool works by sending traffic to the specified target IP and p
     echo $SVC_IP
     ```
 
-1. Then update the `<target-ip>` in `./config/manifests/benchmark/benchmark.yaml` to your target IP. Feel free to adjust other parameters such as request_rates as well. For a complete list of LPG configurations, pls refer to the [LPG user guide](https://github.com/AI-Hypercomputer/inference-benchmark?tab=readme-ov-file#configuring-the-benchmark).
+1. Then update the `<target-ip>` in `./config/manifests/benchmark/benchmark.yaml` to the value of `$SVC_IP` or `$GW_IP`.
+   Feel free to adjust other parameters such as `request_rates` as well. For a complete list of LPG configurations, refer to the
+   [LPG user guide](https://github.com/AI-Hypercomputer/inference-benchmark?tab=readme-ov-file#configuring-the-benchmark).
 
-1. Start the benchmark tool. `kubectl apply -f ./config/manifests/benchmark/benchmark.yaml`
+1. Start the benchmark tool.
 
-1. Wait for benchmark to finish and download the results. Use the `benchmark_id` environment variable
-to specify what this benchmark is for. For instance, `inference-extension` or `k8s-svc`. When the LPG tool finishes benchmarking, it will print a log line `LPG_FINISHED`,
-the script below will watch for that log line and then start downloading results.
+    ```bash
+    kubectl apply -f ./config/manifests/benchmark/benchmark.yaml
+    ```
+
+1. Wait for benchmark to finish and download the results. Use the `benchmark_id` environment variable to specify what this
+   benchmark is for. For instance, `inference-extension` or `k8s-svc`. When the LPG tool finishes benchmarking, it will print
+   a log line `LPG_FINISHED`. The script below will watch for that log line and then start downloading results.
 
     ```bash
-    benchmark_id='my-benchmark' ./tools/benchmark/download-benchmark-results.bash
+    benchmark_id='k8s-svc' ./tools/benchmark/download-benchmark-results.bash
     ```
-    1. After the script finishes, you should see benchmark results under `./tools/benchmark/output/default-run/my-benchmark/results/json` folder. Here is a [sample json file](./sample.json).
+
+    After the script finishes, you should see benchmark results under `./tools/benchmark/output/default-run/k8s-svc/results/json` folder.
+    Here is a [sample json file](./sample.json). Replace `k8s-svc` with `inference-extension` when running an inference extension benchmark.
 
 ### Tips
 
+* When using a `benchmark_id` other than `k8s-svc` or `inference-extension`, the labels in `./tools/benchmark/benchmark.ipynb` must be
+  updated accordingly to analyze the results.
 * You can specify `run_id="runX"` environment variable when running the `./download-benchmark-results.bash` script.
 This is useful when you run benchmarks multiple times to get a more statistically meaningful results and group the results accordingly.
 * Update the `request_rates` that best suit your benchmark environment.
 
 ### Advanced Benchmark Configurations
 
-Pls refer to the [LPG user guide](https://github.com/AI-Hypercomputer/inference-benchmark?tab=readme-ov-file#configuring-the-benchmark) for a detailed list of configuration knobs.
+Refer to the [LPG user guide](https://github.com/AI-Hypercomputer/inference-benchmark?tab=readme-ov-file#configuring-the-benchmark) for a
+detailed list of configuration knobs.
 
 ## Analyze the results
 
-This guide shows how to run the jupyter notebook using vscode.
+This guide shows how to run the jupyter notebook using vscode after completing k8s service and inference extension benchmarks.
 
 1. Create a python virtual environment.
 
@@ -92,6 +107,6 @@ This guide shows how to run the jupyter notebook using vscode.
     ```
 
 1. Open the notebook `./tools/benchmark/benchmark.ipynb`, and run each cell. At the end you should
-    see a bar chart like below where **"ie"** represents inference extension. This chart is generated using this benchmarking tool with 6 vLLM (v1) model servers (H100 80 GB), [llama2-7b](https://huggingface.co/meta-llama/Llama-2-7b-chat-hf/tree/main) and the [ShareGPT dataset](https://huggingface.co/datasets/anon8231489123/ShareGPT_Vicuna_unfiltered/resolve/main/ShareGPT_V3_unfiltered_cleaned_split.json).
-    
-    ![alt text](example-bar-chart.png)
\ No newline at end of file
+   see a bar chart like below where __"ie"__ represents inference extension. This chart is generated using this benchmarking tool with 6 vLLM (v1) model servers (H100 80 GB), [llama2-7b](https://huggingface.co/meta-llama/Llama-2-7b-chat-hf/tree/main) and the [ShareGPT dataset](https://huggingface.co/datasets/anon8231489123/ShareGPT_Vicuna_unfiltered/resolve/main/ShareGPT_V3_unfiltered_cleaned_split.json).
+
+    ![alt text](example-bar-chart.png)
diff --git a/tools/benchmark/benchmark.ipynb b/tools/benchmark/benchmark.ipynb
index 993279c..ffd4c45 100644
--- a/tools/benchmark/benchmark.ipynb
+++ b/tools/benchmark/benchmark.ipynb
@@ -21,7 +21,7 @@
         "#@title Configuration. Edit this before running the rest.\n",
         "\n",
         "OUTPUT_DIR='output'\n",
-        "RUN_ID='example-run'\n",
+        "RUN_ID='default-run'\n",
         "# Path to the benchmark dir under `gateway-api-inference-extension/benchmark`\n",
         "BENCHMARK_DIR =\"./\"\n",
         "# A regex to match the model name, which matches the output file name.\n",
@@ -229,7 +229,7 @@
         "    plot_func(curAx, m)\n",
         "  return fig, axes\n",
         "\n",
-        "def plot_bar(labels, groups, metrics=CORE_METRICS, num_plots_per_row=NUM_PLOTS_PER_ROW, interactive=INTERACTIVE_PLOT, annotate=False):\n",
+        "def plot_bar(labels, groups, metrics=CORE_METRICS, num_plots_per_row=NUM_PLOTS_PER_ROW, interactive=False, annotate=False):\n",
         "    labels = [label.alias for label in labels]\n",
         "    logger.debug(f'Prnting bar chart for {labels}')\n",
         "    logger.debug(f'groups: {groups}')\n",



Test patch:
diff --git a/conformance/tests/basic/inferencepool_accepted.yaml b/conformance/tests/basic/inferencepool_accepted.yaml
index 8ae327d..ecee5b3 100644
--- a/conformance/tests/basic/inferencepool_accepted.yaml
+++ b/conformance/tests/basic/inferencepool_accepted.yaml
@@ -1,8 +1,46 @@
 # Basic InferencePool for acceptance testing.
-# This manifest defines the minimal required fields to create a valid
-# InferencePool resource, which the InferencePoolAccepted test will use
-# to verify that the controller recognizes and accepts the resource.
+# This manifest defines the minimal required fields to create valid
+# InferencePool and HTTPRoute resources, which the InferencePoolAccepted
+# test will use to verify that the controller recognizes and accepts the resource.
 
+# --- Minimal Backend Deployment (using agnhost echo server) ---
+# This Deployment provides Pods for the InferencePool to select.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: infra-backend-v1-deployment
+  namespace: gateway-conformance-app-backend
+  labels:
+    app: infra-backend-v1
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: infra-backend-v1
+  template:
+    metadata:
+      labels:
+        app: infra-backend-v1
+    spec:
+      containers:
+      - name: agnhost-echo
+        image: k8s.gcr.io/e2e-test-images/agnhost:2.39
+        args:
+        - serve-hostname
+        - --http-port=8080
+        ports:
+        - name: http
+          containerPort: 8080
+        readinessProbe:
+          httpGet:
+            path: /
+            port: 8080
+          initialDelaySeconds: 3
+          periodSeconds: 5
+          failureThreshold: 2
+
+---
+# --- InferencePool Definition ---
 apiVersion: inference.networking.x-k8s.io/v1alpha2
 kind: InferencePool
 metadata:
@@ -18,10 +56,38 @@ spec:
     app: "infra-backend-v1"
 
   # --- Target Port (Required) ---
-  # The port the model server container listens on.
-  targetPortNumber: 3000
+  # The port the model server container (agnhost in this case) listens on.
+  targetPortNumber: 8080 # Matches agnhost's http-port
 
   # --- Extension Reference ---
   # GKE-specific configuration reference.
   extensionRef:
+    # group: "" # Optional
+    # kind: Service # Optional
     name: infra-backend-v1-epp
+
+---
+# --- HTTPRoute Definition ---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: httproute-for-inferencepool-accepted
+  namespace: gateway-conformance-app-backend
+spec:
+  parentRefs:
+  - group: gateway.networking.k8s.io
+    kind: Gateway
+    name: conformance-gateway # Name of the shared Gateway from maniffests.yaml
+    namespace: gateway-conformance-infra  # Namespace of the shared Gateway
+    sectionName: http
+  rules:
+  - backendRefs:
+    - group: inference.networking.x-k8s.io # InferencePool API group
+      kind: InferencePool
+      name: inferencepool-basic-accepted # Name of the InferencePool this route points to
+      # namespace: gateway-conformance-app-backend - is omitted since it is in the same namespace as HTTPRoute
+      port: 8080 # Matching the InferencePool's targetPortNumber
+    matches:
+    - path:
+        type: PathPrefix
+        value: /accepted-pool-test
diff --git a/pkg/epp/handlers/server_test.go b/pkg/epp/handlers/server_test.go
deleted file mode 100644
index 23d2b68..0000000
--- a/pkg/epp/handlers/server_test.go
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
-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 handlers
-
-import (
-	"testing"
-	"time"
-
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
-	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
-)
-
-func TestRandomWeightedDraw(t *testing.T) {
-	logger := logutil.NewTestLogger()
-	tests := []struct {
-		name  string
-		model *v1alpha2.InferenceModel
-		want  string
-	}{
-		{
-			name: "'random' distribution",
-			model: &v1alpha2.InferenceModel{
-				Spec: v1alpha2.InferenceModelSpec{
-					TargetModels: []v1alpha2.TargetModel{
-						{
-							Name:   "canary",
-							Weight: pointer(50),
-						},
-						{
-							Name:   "v1",
-							Weight: pointer(50),
-						},
-					},
-				},
-			},
-			want: "canary",
-		},
-		{
-			name: "'random' distribution",
-			model: &v1alpha2.InferenceModel{
-				Spec: v1alpha2.InferenceModelSpec{
-					TargetModels: []v1alpha2.TargetModel{
-						{
-							Name:   "canary",
-							Weight: pointer(25),
-						},
-						{
-							Name:   "v1.1",
-							Weight: pointer(55),
-						},
-						{
-							Name:   "v1",
-							Weight: pointer(50),
-						},
-					},
-				},
-			},
-			want: "v1",
-		},
-		{
-			name: "'random' distribution",
-			model: &v1alpha2.InferenceModel{
-				Spec: v1alpha2.InferenceModelSpec{
-					TargetModels: []v1alpha2.TargetModel{
-						{
-							Name:   "canary",
-							Weight: pointer(20),
-						},
-						{
-							Name:   "v1.1",
-							Weight: pointer(20),
-						},
-						{
-							Name:   "v1",
-							Weight: pointer(10),
-						},
-					},
-				},
-			},
-			want: "v1.1",
-		},
-		{
-			name: "weighted distribution with weight unset",
-			model: &v1alpha2.InferenceModel{
-				Spec: v1alpha2.InferenceModelSpec{
-					TargetModels: []v1alpha2.TargetModel{
-						{
-							Name: "canary",
-						},
-						{
-							Name: "v1.1",
-						},
-						{
-							Name: "v1",
-						},
-					},
-				},
-			},
-			want: "canary",
-		},
-	}
-	var seedVal int64 = 420
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			for range 10000 {
-				model := RandomWeightedDraw(logger, test.model, seedVal)
-				if model != test.want {
-					t.Errorf("Model returned: %v != %v", model, test.want)
-					break
-				}
-			}
-		})
-	}
-}
-
-func TestGetRandomPod(t *testing.T) {
-	tests := []struct {
-		name      string
-		storePods []*corev1.Pod
-		expectNil bool
-	}{
-		{
-			name:      "No pods available",
-			storePods: []*corev1.Pod{},
-			expectNil: true,
-		},
-		{
-			name: "Single pod available",
-			storePods: []*corev1.Pod{
-				{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}},
-			},
-			expectNil: false,
-		},
-		{
-			name: "Multiple pods available",
-			storePods: []*corev1.Pod{
-				{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}},
-				{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}},
-				{ObjectMeta: metav1.ObjectMeta{Name: "pod3"}},
-			},
-			expectNil: false,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			pmf := metrics.NewPodMetricsFactory(&metrics.FakePodMetricsClient{}, time.Millisecond)
-			ds := datastore.NewDatastore(t.Context(), pmf)
-			for _, pod := range test.storePods {
-				ds.PodUpdateOrAddIfNotExist(pod)
-			}
-
-			gotPod := GetRandomPod(ds)
-
-			if test.expectNil && gotPod != nil {
-				t.Errorf("expected nil pod, got: %v", gotPod)
-			}
-			if !test.expectNil && gotPod == nil {
-				t.Errorf("expected non-nil pod, got nil")
-			}
-		})
-	}
-}
-
-func pointer(v int32) *int32 {
-	return &v
-}
diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go
index 3a8136a..4ad6f96 100644
--- a/pkg/epp/metrics/metrics_test.go
+++ b/pkg/epp/metrics/metrics_test.go
@@ -664,3 +664,106 @@ func TestSchedulerE2ELatency(t *testing.T) {
 		})
 	}
 }
+
+func TestPrefixCacheMetrics(t *testing.T) {
+	const (
+		PrefixCacheSizeMetric      = InferenceExtension + "_prefix_indexer_size"
+		PrefixCacheHitRatioMetric  = InferenceExtension + "_prefix_indexer_hit_ratio"
+		PrefixCacheHitLengthMetric = InferenceExtension + "_prefix_indexer_hit_bytes"
+	)
+
+	type cacheMatchRecord struct {
+		matchedLength int
+		totalLength   int
+	}
+
+	scenario := struct {
+		name         string
+		cacheSizes   []int64
+		cacheMatches []cacheMatchRecord
+	}{
+		name:       "multiple cache metrics",
+		cacheSizes: []int64{1024, 2048, 4096},
+		cacheMatches: []cacheMatchRecord{
+			{
+				matchedLength: 5,
+				totalLength:   10,
+			},
+			{
+				matchedLength: 0,
+				totalLength:   10,
+			},
+			{
+				matchedLength: 10,
+				totalLength:   10,
+			},
+			{
+				matchedLength: 7,
+				totalLength:   10,
+			},
+			{
+				matchedLength: 64,
+				totalLength:   128,
+			},
+			{
+				matchedLength: 0,
+				totalLength:   128,
+			},
+		},
+	}
+
+	Register()
+	t.Run(scenario.name, func(t *testing.T) {
+		// Record cache size metrics
+		for _, size := range scenario.cacheSizes {
+			RecordPrefixCacheSize(size)
+		}
+
+		// Record cache match metrics (both hit ratio and hit length)
+		for _, match := range scenario.cacheMatches {
+			RecordPrefixCacheMatch(match.matchedLength, match.totalLength)
+		}
+
+		// Verify cache size metrics
+		wantCacheSizeMetrics, err := os.Open("testdata/prefix_indexer_size_metric")
+		defer func() {
+			if err := wantCacheSizeMetrics.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, wantCacheSizeMetrics, PrefixCacheSizeMetric); err != nil {
+			t.Error(err)
+		}
+
+		// Verify hit ratio metrics
+		wantHitRatioMetrics, err := os.Open("testdata/prefix_indexer_hit_ratio_metric")
+		defer func() {
+			if err := wantHitRatioMetrics.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, wantHitRatioMetrics, PrefixCacheHitRatioMetric); err != nil {
+			t.Error(err)
+		}
+
+		// Verify hit length metrics
+		wantHitLengthMetrics, err := os.Open("testdata/prefix_indexer_hit_bytes_metric")
+		defer func() {
+			if err := wantHitLengthMetrics.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, wantHitLengthMetrics, PrefixCacheHitLengthMetric); err != nil {
+			t.Error(err)
+		}
+	})
+}
diff --git a/pkg/epp/metrics/testdata/prefix_indexer_hit_bytes_metric b/pkg/epp/metrics/testdata/prefix_indexer_hit_bytes_metric
new file mode 100644
index 0000000..86b4872
--- /dev/null
+++ b/pkg/epp/metrics/testdata/prefix_indexer_hit_bytes_metric
@@ -0,0 +1,19 @@
+# HELP inference_extension_prefix_indexer_hit_bytes [ALPHA] Length of the prefix match in number of bytes in the cache lookup.
+# TYPE inference_extension_prefix_indexer_hit_bytes histogram
+inference_extension_prefix_indexer_hit_bytes_bucket{le="0"} 2
+inference_extension_prefix_indexer_hit_bytes_bucket{le="16"} 5
+inference_extension_prefix_indexer_hit_bytes_bucket{le="32"} 5
+inference_extension_prefix_indexer_hit_bytes_bucket{le="64"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="128"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="256"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="512"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="1024"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="2048"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="4096"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="8192"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="16384"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="32768"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="65536"} 6
+inference_extension_prefix_indexer_hit_bytes_bucket{le="+Inf"} 6
+inference_extension_prefix_indexer_hit_bytes_sum 86
+inference_extension_prefix_indexer_hit_bytes_count 6
diff --git a/pkg/epp/metrics/testdata/prefix_indexer_hit_ratio_metric b/pkg/epp/metrics/testdata/prefix_indexer_hit_ratio_metric
new file mode 100644
index 0000000..e94827c
--- /dev/null
+++ b/pkg/epp/metrics/testdata/prefix_indexer_hit_ratio_metric
@@ -0,0 +1,16 @@
+# HELP inference_extension_prefix_indexer_hit_ratio [ALPHA] Ratio of prefix length matched to total prefix length in the cache lookup.
+# TYPE inference_extension_prefix_indexer_hit_ratio histogram
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0"} 2
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.1"} 2
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.2"} 2
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.3"} 2
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.4"} 2
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.5"} 4
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.6"} 4
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.7"} 5
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.8"} 5
+inference_extension_prefix_indexer_hit_ratio_bucket{le="0.9"} 5
+inference_extension_prefix_indexer_hit_ratio_bucket{le="1"} 6
+inference_extension_prefix_indexer_hit_ratio_bucket{le="+Inf"} 6
+inference_extension_prefix_indexer_hit_ratio_sum 2.7
+inference_extension_prefix_indexer_hit_ratio_count 6
diff --git a/pkg/epp/metrics/testdata/prefix_indexer_size_metric b/pkg/epp/metrics/testdata/prefix_indexer_size_metric
new file mode 100644
index 0000000..9799b17
--- /dev/null
+++ b/pkg/epp/metrics/testdata/prefix_indexer_size_metric
@@ -0,0 +1,3 @@
+# HELP inference_extension_prefix_indexer_size [ALPHA] Size of the prefix indexer.
+# TYPE inference_extension_prefix_indexer_size gauge
+inference_extension_prefix_indexer_size{} 4096
diff --git a/pkg/epp/handlers/request_test.go b/pkg/epp/requestcontrol/director_test.go
similarity index 59%
rename from pkg/epp/handlers/request_test.go
rename to pkg/epp/requestcontrol/director_test.go
index 8675b64..05dc1b3 100644
--- a/pkg/epp/handlers/request_test.go
+++ b/pkg/epp/requestcontrol/director_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package handlers
+package requestcontrol
 
 import (
 	"context"
@@ -23,26 +23,25 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
 	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 	testutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/testing"
 )
 
-const (
-	DefaultDestinationEndpointHintMetadataNamespace = "envoy.lb"                       // default for --destinationEndpointHintMetadataNamespace
-	DefaultDestinationEndpointHintKey               = "x-gateway-destination-endpoint" // default for --destinationEndpointHintKey
-)
-
-func TestHandleRequestBody(t *testing.T) {
+func TestHandleRequest(t *testing.T) {
 	ctx := logutil.NewTestLoggerIntoContext(context.Background())
 
 	// Setup datastore
@@ -82,7 +81,7 @@ func TestHandleRequestBody(t *testing.T) {
 		name         string
 		reqBodyMap   map[string]interface{}
 		wantErrCode  string
-		wantReqCtx   *RequestContext
+		wantReqCtx   *handlers.RequestContext
 		wantRespBody map[string]interface{}
 	}{
 		{
@@ -91,7 +90,7 @@ func TestHandleRequestBody(t *testing.T) {
 				"model":  tsModel,
 				"prompt": "test prompt",
 			},
-			wantReqCtx: &RequestContext{
+			wantReqCtx: &handlers.RequestContext{
 				Model:               tsModel,
 				ResolvedTargetModel: tsModel,
 				TargetPod:           "/pod1",
@@ -108,7 +107,7 @@ func TestHandleRequestBody(t *testing.T) {
 				"model":  modelWithTarget,
 				"prompt": "test prompt",
 			},
-			wantReqCtx: &RequestContext{
+			wantReqCtx: &handlers.RequestContext{
 				Model:               modelWithTarget,
 				ResolvedTargetModel: modelWithTarget,
 				TargetPod:           "/pod1",
@@ -143,13 +142,13 @@ func TestHandleRequestBody(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			server := NewStreamingServer(scheduling.NewScheduler(ds), DefaultDestinationEndpointHintMetadataNamespace, DefaultDestinationEndpointHintKey, ds)
-			reqCtx := &RequestContext{
-				Request: &Request{
+			server := NewDirector(ds, scheduling.NewScheduler(ds))
+			reqCtx := &handlers.RequestContext{
+				Request: &handlers.Request{
 					Body: test.reqBodyMap,
 				},
 			}
-			reqCtx, err := server.HandleRequestBody(ctx, reqCtx)
+			reqCtx, err := server.HandleRequest(ctx, reqCtx)
 
 			if test.wantErrCode != "" {
 				if err == nil {
@@ -182,3 +181,159 @@ func TestHandleRequestBody(t *testing.T) {
 		})
 	}
 }
+
+func TestRandomWeightedDraw(t *testing.T) {
+	logger := logutil.NewTestLogger()
+	tests := []struct {
+		name  string
+		model *v1alpha2.InferenceModel
+		want  string
+	}{
+		{
+			name: "'random' distribution",
+			model: &v1alpha2.InferenceModel{
+				Spec: v1alpha2.InferenceModelSpec{
+					TargetModels: []v1alpha2.TargetModel{
+						{
+							Name:   "canary",
+							Weight: pointer(50),
+						},
+						{
+							Name:   "v1",
+							Weight: pointer(50),
+						},
+					},
+				},
+			},
+			want: "canary",
+		},
+		{
+			name: "'random' distribution",
+			model: &v1alpha2.InferenceModel{
+				Spec: v1alpha2.InferenceModelSpec{
+					TargetModels: []v1alpha2.TargetModel{
+						{
+							Name:   "canary",
+							Weight: pointer(25),
+						},
+						{
+							Name:   "v1.1",
+							Weight: pointer(55),
+						},
+						{
+							Name:   "v1",
+							Weight: pointer(50),
+						},
+					},
+				},
+			},
+			want: "v1",
+		},
+		{
+			name: "'random' distribution",
+			model: &v1alpha2.InferenceModel{
+				Spec: v1alpha2.InferenceModelSpec{
+					TargetModels: []v1alpha2.TargetModel{
+						{
+							Name:   "canary",
+							Weight: pointer(20),
+						},
+						{
+							Name:   "v1.1",
+							Weight: pointer(20),
+						},
+						{
+							Name:   "v1",
+							Weight: pointer(10),
+						},
+					},
+				},
+			},
+			want: "v1.1",
+		},
+		{
+			name: "weighted distribution with weight unset",
+			model: &v1alpha2.InferenceModel{
+				Spec: v1alpha2.InferenceModelSpec{
+					TargetModels: []v1alpha2.TargetModel{
+						{
+							Name: "canary",
+						},
+						{
+							Name: "v1.1",
+						},
+						{
+							Name: "v1",
+						},
+					},
+				},
+			},
+			want: "canary",
+		},
+	}
+	var seedVal int64 = 420
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			for range 10000 {
+				model := RandomWeightedDraw(logger, test.model, seedVal)
+				if model != test.want {
+					t.Errorf("Model returned: %v != %v", model, test.want)
+					break
+				}
+			}
+		})
+	}
+}
+
+func TestGetRandomPod(t *testing.T) {
+	tests := []struct {
+		name      string
+		storePods []*corev1.Pod
+		expectNil bool
+	}{
+		{
+			name:      "No pods available",
+			storePods: []*corev1.Pod{},
+			expectNil: true,
+		},
+		{
+			name: "Single pod available",
+			storePods: []*corev1.Pod{
+				{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}},
+			},
+			expectNil: false,
+		},
+		{
+			name: "Multiple pods available",
+			storePods: []*corev1.Pod{
+				{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "pod3"}},
+			},
+			expectNil: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			pmf := metrics.NewPodMetricsFactory(&metrics.FakePodMetricsClient{}, time.Millisecond)
+			ds := datastore.NewDatastore(t.Context(), pmf)
+			for _, pod := range test.storePods {
+				ds.PodUpdateOrAddIfNotExist(pod)
+			}
+			d := &Director{datastore: ds}
+			gotPod := d.GetRandomPod()
+
+			if test.expectNil && gotPod != nil {
+				t.Errorf("expected nil pod, got: %v", gotPod)
+			}
+			if !test.expectNil && gotPod == nil {
+				t.Errorf("expected non-nil pod, got nil")
+			}
+		})
+	}
+}
+
+func pointer(v int32) *int32 {
+	return &v
+}
diff --git a/pkg/epp/scheduling/plugins/filter/filter_test.go b/pkg/epp/scheduling/plugins/filter/filter_test.go
index 2354c3e..1737f61 100644
--- a/pkg/epp/scheduling/plugins/filter/filter_test.go
+++ b/pkg/epp/scheduling/plugins/filter/filter_test.go
@@ -21,64 +21,47 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/config"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
-func TestFilter(t *testing.T) {
-	tests := []struct {
-		name   string
-		req    *types.LLMRequest
-		input  []types.Pod
-		output []types.Pod
-		filter *DecisionTreeFilter
-	}{
-		{
-			name: "simple filter without available pods",
-			filter: &DecisionTreeFilter{
-				Current: &baseFilter{
-					name: "filter all",
-					filter: func(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-						return []types.Pod{}
-					},
-				},
-			},
-			output: []types.Pod{},
-		},
-	}
+type filterAll struct{}
 
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			ctx := types.NewSchedulingContext(context.Background(), test.req, test.input)
-			got := test.filter.Filter(ctx, test.input)
+func (f *filterAll) Name() string {
+	return "filter all"
+}
 
-			if diff := cmp.Diff(test.output, got); diff != "" {
-				t.Errorf("Unexpected output (-want +got): %v", diff)
-			}
-		})
-	}
+func (f *filterAll) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	return []types.Pod{}
 }
 
-func TestFilterFunc(t *testing.T) {
+func TestFilter(t *testing.T) {
 	tests := []struct {
 		name   string
-		f      filterFunc
 		req    *types.LLMRequest
+		filter plugins.Filter
 		input  []types.Pod
 		output []types.Pod
 	}{
+		{
+			name:   "simple filter filters all pods",
+			filter: &filterAll{},
+			output: []types.Pod{},
+		},
 		{
 			name:   "least queuing empty input",
-			f:      leastQueuingFilterFunc,
+			filter: NewLeastQueueFilter(),
 			input:  []types.Pod{},
 			output: []types.Pod{},
 		},
 		{
-			name: "least queuing",
-			f:    leastQueuingFilterFunc,
+			name:   "least queuing",
+			filter: NewLeastQueueFilter(),
 			input: []types.Pod{
 				&types.PodMetrics{
 					Metrics: &backendmetrics.Metrics{
@@ -111,13 +94,13 @@ func TestFilterFunc(t *testing.T) {
 		},
 		{
 			name:   "least kv cache empty input",
-			f:      leastKVCacheFilterFunc,
+			filter: NewLeastKVCacheFilter(),
 			input:  []types.Pod{},
 			output: []types.Pod{},
 		},
 		{
-			name: "least kv cache",
-			f:    leastKVCacheFilterFunc,
+			name:   "least kv cache",
+			filter: NewLeastKVCacheFilter(),
 			input: []types.Pod{
 				&types.PodMetrics{
 					Metrics: &backendmetrics.Metrics{
@@ -149,8 +132,9 @@ func TestFilterFunc(t *testing.T) {
 			},
 		},
 		{
-			name: "lowQueueAndLessThanKVCacheThresholdPredicate",
-			f:    toFilterFunc(queueThresholdPredicate(0).and(kvCacheThresholdPredicate(0.8))),
+			name:   "SheddableCapacityFilter, sheddable request",
+			req:    &types.LLMRequest{Critical: false},
+			filter: &SheddableCapacityFilter{queueThreshold: 0, kvCacheThreshold: 0.8},
 			input: []types.Pod{
 				&types.PodMetrics{
 					// This pod should be returned.
@@ -187,8 +171,8 @@ func TestFilterFunc(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			ctx := types.NewSchedulingContext(context.Background(), test.req, test.input)
-			got := test.f(ctx, test.input)
+			ctx := types.NewSchedulingContext(context.Background(), test.req, nil, test.input)
+			got := test.filter.Filter(ctx, test.input)
 
 			if diff := cmp.Diff(test.output, got); diff != "" {
 				t.Errorf("Unexpected output (-want +got): %v", diff)
@@ -221,8 +205,8 @@ func TestLoRASoftAffinityDistribution(t *testing.T) {
 
 	// Create a test request and pods
 	req := &types.LLMRequest{
-		Model:               testAffinityModel,
-		ResolvedTargetModel: testAffinityModel,
+		TargetModel: testAffinityModel,
+		RequestId:   uuid.NewString(),
 	}
 
 	// Test setup: One affinity pod and one available pod
@@ -244,7 +228,7 @@ func TestLoRASoftAffinityDistribution(t *testing.T) {
 			},
 		},
 	}
-	ctx := types.NewSchedulingContext(context.Background(), req, pods)
+	ctx := types.NewSchedulingContext(context.Background(), req, nil, pods)
 
 	// Run the filter function multiple times and count the results
 	affinityCount := 0
@@ -254,8 +238,11 @@ func TestLoRASoftAffinityDistribution(t *testing.T) {
 	expectedAffinityPercent := config.Conf.LoraAffinityThreshold * 100
 	expectedAvailabilityPercent := 100 - expectedAffinityPercent
 
+	// initialize LoraAffinityFilter
+	LoraAffinityFilter := NewLoraAffinityFilter()
+
 	for i := 0; i < numIterations; i++ {
-		result := loRASoftAffinityFilterFunc(ctx, pods)
+		result := LoraAffinityFilter.Filter(ctx, pods)
 
 		// Check which type of pod was returned
 		if len(result) != 1 {
diff --git a/pkg/epp/scheduling/plugins/prefix/indexer_test.go b/pkg/epp/scheduling/plugins/prefix/indexer_test.go
new file mode 100644
index 0000000..596625d
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/prefix/indexer_test.go
@@ -0,0 +1,45 @@
+/*
+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 prefix
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIndexer_AddAndGet(t *testing.T) {
+	cache := newIndexer(2)
+
+	hash1 := BlockHash(1)
+	server := ServerID{Namespace: "default", Name: "server1"}
+
+	// Add an entry to the cache
+	cache.Add([]BlockHash{hash1}, server)
+
+	// Retrieve the entry
+	assert.Equal(t, 1, cache.ll.Len(), "Cache size should be 1 after adding an entry")
+	servers := cache.Get(hash1)
+	assert.Contains(t, servers, server, "Cache should contain the added server")
+
+	// Add another entry to the cache, the cache size should be incremented to 2.
+	cache.Add([]BlockHash{BlockHash(2)}, server)
+	assert.Equal(t, 2, cache.ll.Len(), "Cache size should  be 2 after adding an entry")
+
+	// Add another entry to the cache, which should evict the first one due to max size.
+	cache.Add([]BlockHash{BlockHash(3)}, server)
+	assert.Equal(t, 2, cache.ll.Len(), "Cache size should still be 2 after adding an entry")
+}
diff --git a/pkg/epp/scheduling/plugins/prefix/plugin_test.go b/pkg/epp/scheduling/plugins/prefix/plugin_test.go
new file mode 100644
index 0000000..7e4e218
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/prefix/plugin_test.go
@@ -0,0 +1,137 @@
+package prefix
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	k8stypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
+)
+
+func TestPrefixPlugin(t *testing.T) {
+	config := Config{
+		HashBlockSize:          4,
+		MaxPrefixBlocksToMatch: DefaultMaxPrefixBlocks,
+		LRUIndexerCapacity:     DefaultLRUIndexerCapacity,
+	}
+	plugin := New(config)
+
+	pod1 := &types.PodMetrics{Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod1"}}}
+	pod2 := &types.PodMetrics{Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod2"}}}
+	pods := []types.Pod{pod1, pod2}
+
+	// First request.
+	req1 := &types.LLMRequest{
+		TargetModel: "test-model1",
+		Prompt:      "aaaaaa",
+	}
+	ctx := types.NewSchedulingContext(context.Background(), req1, nil, pods)
+	plugin.PreSchedule(ctx)
+	state, err := plugin.getPrefixState(ctx.CycleState)
+	assert.NoError(t, err)
+	t.Logf("Hashes %+v, cached servers: %+v", state.PrefixHashes, state.PrefixCacheServers)
+	// Input size is 6, hash block size is 4, the last 2 characters are ignored.
+	// Total hashes = 2 (the first one is for the model)
+	assert.Equal(t, 2, len(state.PrefixHashes), "number of hashes is incorrect")
+	assert.Equal(t, 0, len(state.PrefixCacheServers), "there shouldn't be any cached servers")
+
+	// Updated to use the new Score method signature
+	scores := plugin.Score(ctx, pods)
+	assert.Equal(t, float64(0), scores[pod1], "score for pod1")
+	assert.Equal(t, float64(0), scores[pod2], "score for pod2")
+
+	// Simulate pod1 was picked.
+	plugin.PostSchedule(ctx, &types.Result{TargetPod: pod1})
+
+	// Second request doesn't share any prefix with first one. It should be added to the cache but
+	// the pod score should be 0.
+	req2 := &types.LLMRequest{
+		TargetModel: "test-model2",
+		Prompt:      "bbbbbb",
+	}
+	ctx = types.NewSchedulingContext(context.Background(), req2, nil, pods)
+	plugin.PreSchedule(ctx)
+	state, err = plugin.getPrefixState(ctx.CycleState)
+	assert.NoError(t, err)
+	t.Logf("Hashes %+v, cached servers: %+v", state.PrefixHashes, state.PrefixCacheServers)
+	// Input size is 6, hash block size is 4, the last 2 characters are ignored.
+	// Total hashes = 2 (the first one is for the model)
+	assert.Equal(t, 2, len(state.PrefixHashes), "number of hashes is incorrect")
+	assert.Equal(t, 0, len(state.PrefixCacheServers), "there shouldn't be any cached servers")
+
+	// Updated to use the new Score method signature
+	scores = plugin.Score(ctx, pods)
+	assert.Equal(t, float64(0), scores[pod1], "score for pod1")
+	assert.Equal(t, float64(0), scores[pod2], "score for pod2")
+
+	// Simulate pod2 was picked.
+	plugin.PostSchedule(ctx, &types.Result{TargetPod: pod2})
+
+	// Third request shares partial prefix with first one.
+	req3 := &types.LLMRequest{
+		TargetModel: "test-model1",
+		Prompt:      "aaaabbbb",
+	}
+	ctx = types.NewSchedulingContext(context.Background(), req3, nil, pods)
+	plugin.PreSchedule(ctx)
+	state, err = plugin.getPrefixState(ctx.CycleState)
+	assert.NoError(t, err)
+	t.Logf("Hashes %+v, cached servers: %+v", state.PrefixHashes, state.PrefixCacheServers)
+	// Input size is 8, hash block size is 4, so 2 hashes will be calculated.
+	// Total hashes = 3 (the first one is for the model)
+	assert.Equal(t, 3, len(state.PrefixHashes), "number of hashes is incorrect")
+	assert.Equal(t, 1, len(state.PrefixCacheServers), "pod1 should have cached the aaaa prefix")
+
+	// Updated to use the new Score method signature
+	scores = plugin.Score(ctx, pods)
+	assert.Equal(t, float64(2)/float64(3), scores[pod1], "score should be 2/3 - the model and the first prefix block match")
+	assert.Equal(t, float64(0), scores[pod2], "score for pod2")
+
+	plugin.PostSchedule(ctx, &types.Result{TargetPod: pod1})
+
+	// 4th request is same as req3 except the model is different, still no match.
+	req4 := &types.LLMRequest{
+		TargetModel: "test-model-new",
+		Prompt:      "aaaabbbb",
+	}
+	ctx = types.NewSchedulingContext(context.Background(), req4, nil, pods)
+	plugin.PreSchedule(ctx)
+	state, err = plugin.getPrefixState(ctx.CycleState)
+	assert.NoError(t, err)
+	t.Logf("Hashes %+v, cached servers: %+v", state.PrefixHashes, state.PrefixCacheServers)
+	// Input size is 8, hash block size is 4, so 2 hashes will be calculated.
+	// Total hashes = 3 (the first one is for the model)
+	assert.Equal(t, 3, len(state.PrefixHashes), "number of hashes is incorrect")
+	assert.Equal(t, 0, len(state.PrefixCacheServers), "pod1 should have cached the aaaa prefix")
+
+	// Updated to use the new Score method signature
+	scores = plugin.Score(ctx, pods)
+	assert.Equal(t, float64(0), scores[pod1], "score for pod1")
+	assert.Equal(t, float64(0), scores[pod2], "score for pod2")
+
+	plugin.PostSchedule(ctx, &types.Result{TargetPod: pod1})
+
+	// 5th request shares partial prefix with 3rd one.
+	req5 := &types.LLMRequest{
+		TargetModel: "test-model1",
+		Prompt:      "aaaabbbbcccc",
+	}
+	ctx = types.NewSchedulingContext(context.Background(), req5, nil, pods)
+	plugin.PreSchedule(ctx)
+	state, err = plugin.getPrefixState(ctx.CycleState)
+	assert.NoError(t, err)
+	t.Logf("Hashes %+v, cached servers: %+v", state.PrefixHashes, state.PrefixCacheServers)
+	// Input size is 12, hash block size is 4, so 3 hashes will be calculated.
+	// Total hashes = 4 (the first one is for the model)
+	assert.Equal(t, 4, len(state.PrefixHashes), "number of hashes is incorrect")
+	assert.Equal(t, 1, len(state.PrefixCacheServers), "pod1 should have cached the aaaa prefix")
+
+	// Updated to use the new Score method signature
+	scores = plugin.Score(ctx, pods)
+	assert.Equal(t, 0.75, scores[pod1], "score should be 0.75 - the model and the first 2 prefix blocks match")
+	assert.Equal(t, float64(0), scores[pod2], "score for pod2")
+
+	plugin.PostSchedule(ctx, &types.Result{TargetPod: pod1})
+}
diff --git a/pkg/epp/scheduling/plugins/scorer/kvcache_test.go b/pkg/epp/scheduling/plugins/scorer/kvcache_test.go
index 257a58c..68be8a2 100644
--- a/pkg/epp/scheduling/plugins/scorer/kvcache_test.go
+++ b/pkg/epp/scheduling/plugins/scorer/kvcache_test.go
@@ -82,7 +82,7 @@ func TestKvCacheScorer(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ctx := types.NewSchedulingContext(context.Background(), &types.LLMRequest{}, tt.pods)
+			ctx := types.NewSchedulingContext(context.Background(), &types.LLMRequest{}, nil, tt.pods)
 			scorer := &KVCacheScorer{}
 			scores := scorer.Score(ctx, tt.pods)
 
diff --git a/pkg/epp/scheduling/plugins/scorer/queue_test.go b/pkg/epp/scheduling/plugins/scorer/queue_test.go
index 907681b..d60eab6 100644
--- a/pkg/epp/scheduling/plugins/scorer/queue_test.go
+++ b/pkg/epp/scheduling/plugins/scorer/queue_test.go
@@ -73,7 +73,7 @@ func TestQueueScorer(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ctx := types.NewSchedulingContext(context.Background(), &types.LLMRequest{}, tt.pods)
+			ctx := types.NewSchedulingContext(context.Background(), &types.LLMRequest{}, nil, tt.pods)
 			scores := scorer.Score(ctx, tt.pods)
 
 			for i, pod := range tt.pods {
diff --git a/pkg/epp/scheduling/scheduler_test.go b/pkg/epp/scheduling/scheduler_test.go
index b44c7ac..8936cc6 100644
--- a/pkg/epp/scheduling/scheduler_test.go
+++ b/pkg/epp/scheduling/scheduler_test.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics" // Import config for thresholds
@@ -40,9 +41,9 @@ func TestSchedule(t *testing.T) {
 		{
 			name: "no pods in datastore",
 			req: &types.LLMRequest{
-				Model:               "any-model",
-				ResolvedTargetModel: "any-model",
-				Critical:            true,
+				TargetModel: "any-model",
+				RequestId:   uuid.NewString(),
+				Critical:    true,
 			},
 			input: []*backendmetrics.FakePodMetrics{},
 			err:   true,
@@ -50,9 +51,9 @@ func TestSchedule(t *testing.T) {
 		{
 			name: "critical request",
 			req: &types.LLMRequest{
-				Model:               "critical",
-				ResolvedTargetModel: "critical",
-				Critical:            true,
+				TargetModel: "critical",
+				RequestId:   uuid.NewString(),
+				Critical:    true,
 			},
 			// pod2 will be picked because it has relatively low queue size, with the requested
 			// model being active, and has low KV cache.
@@ -96,7 +97,7 @@ func TestSchedule(t *testing.T) {
 			wantRes: &types.Result{
 				TargetPod: &types.ScoredPod{
 					Pod: &types.PodMetrics{
-						Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod2"}},
+						Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod2"}, Labels: make(map[string]string)},
 						Metrics: &backendmetrics.Metrics{
 							WaitingQueueSize:    3,
 							KVCacheUsagePercent: 0.1,
@@ -114,9 +115,9 @@ func TestSchedule(t *testing.T) {
 		{
 			name: "sheddable request, accepted",
 			req: &types.LLMRequest{
-				Model:               "sheddable",
-				ResolvedTargetModel: "sheddable",
-				Critical:            false,
+				TargetModel: "sheddable",
+				RequestId:   uuid.NewString(),
+				Critical:    false,
 			},
 			// pod1 will be picked because it has capacity for the sheddable request.
 			input: []*backendmetrics.FakePodMetrics{
@@ -159,7 +160,7 @@ func TestSchedule(t *testing.T) {
 			wantRes: &types.Result{
 				TargetPod: &types.ScoredPod{
 					Pod: &types.PodMetrics{
-						Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod1"}},
+						Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod1"}, Labels: make(map[string]string)},
 						Metrics: &backendmetrics.Metrics{
 							WaitingQueueSize:    0,
 							KVCacheUsagePercent: 0.2,
@@ -177,9 +178,9 @@ func TestSchedule(t *testing.T) {
 		{
 			name: "sheddable request, dropped",
 			req: &types.LLMRequest{
-				Model:               "sheddable",
-				ResolvedTargetModel: "sheddable",
-				Critical:            false,
+				TargetModel: "sheddable",
+				RequestId:   uuid.NewString(),
+				Critical:    false,
 			},
 			// All pods have higher KV cache thant the threshold, so the sheddable request will be
 			// dropped.
@@ -356,7 +357,10 @@ func TestSchedulePlugins(t *testing.T) {
 			// Initialize the scheduler
 			scheduler := NewSchedulerWithConfig(&fakeDataStore{pods: test.input}, &test.config)
 
-			req := &types.LLMRequest{Model: "test-model"}
+			req := &types.LLMRequest{
+				TargetModel: "test-model",
+				RequestId:   uuid.NewString(),
+			}
 			got, err := scheduler.Schedule(context.Background(), req)
 
 			// Validate error state
@@ -370,7 +374,7 @@ func TestSchedulePlugins(t *testing.T) {
 
 			// Validate output
 			wantPod := &types.PodMetrics{
-				Pod: &backend.Pod{NamespacedName: test.wantTargetPod},
+				Pod: &backend.Pod{NamespacedName: test.wantTargetPod, Labels: make(map[string]string)},
 			}
 			wantRes := &types.Result{TargetPod: wantPod}
 			if diff := cmp.Diff(wantRes, got); diff != "" {
@@ -423,6 +427,59 @@ func TestSchedulePlugins(t *testing.T) {
 	}
 }
 
+func TestPostResponse(t *testing.T) {
+	pr1 := &testPostResponse{
+		NameRes:                 "pr1",
+		ExtraHeaders:            map[string]string{"x-session-id": "qwer-asdf-zxcv"},
+		ReceivedResponseHeaders: make(map[string]string),
+	}
+
+	targetPod := k8stypes.NamespacedName{Name: "pod2"}
+
+	tests := []struct {
+		name               string
+		config             SchedulerConfig
+		input              []*backendmetrics.FakePodMetrics
+		responseHeaders    map[string]string
+		wantUpdatedHeaders map[string]string
+	}{
+		{
+			name: "Simple postResponse test",
+			config: SchedulerConfig{
+				postResponsePlugins: []plugins.PostResponse{pr1},
+			},
+			input: []*backendmetrics.FakePodMetrics{
+				{Pod: &backend.Pod{NamespacedName: k8stypes.NamespacedName{Name: "pod1"}}},
+				{Pod: &backend.Pod{NamespacedName: targetPod}},
+			},
+			responseHeaders:    map[string]string{"Content-type": "application/json", "Content-Length": "1234"},
+			wantUpdatedHeaders: map[string]string{"x-session-id": "qwer-asdf-zxcv", "Content-type": "application/json", "Content-Length": "1234"},
+		},
+	}
+
+	for _, test := range tests {
+		scheduler := NewSchedulerWithConfig(&fakeDataStore{pods: test.input}, &test.config)
+
+		headers := map[string]string{}
+		for k, v := range test.responseHeaders {
+			headers[k] = v
+		}
+		resp := &types.LLMResponse{
+			Headers: headers,
+		}
+
+		scheduler.OnResponse(context.Background(), resp, targetPod.String())
+
+		if diff := cmp.Diff(test.responseHeaders, pr1.ReceivedResponseHeaders); diff != "" {
+			t.Errorf("Unexpected output (-responseHeaders +ReceivedResponseHeaders): %v", diff)
+		}
+
+		if diff := cmp.Diff(test.wantUpdatedHeaders, resp.Headers); diff != "" {
+			t.Errorf("Unexpected output (-wantUpdatedHeaders +resp.Headers): %v", diff)
+		}
+	}
+}
+
 type fakeDataStore struct {
 	pods []*backendmetrics.FakePodMetrics
 }
@@ -495,6 +552,23 @@ func (tp *TestPlugin) reset() {
 	tp.NumOfPickerCandidates = 0
 }
 
+type testPostResponse struct {
+	NameRes                 string
+	ReceivedResponseHeaders map[string]string
+	ExtraHeaders            map[string]string
+}
+
+func (pr *testPostResponse) Name() string { return pr.NameRes }
+
+func (pr *testPostResponse) PostResponse(ctx *types.SchedulingContext, pod types.Pod) {
+	for key, value := range ctx.Resp.Headers {
+		pr.ReceivedResponseHeaders[key] = value
+	}
+	for key, value := range pr.ExtraHeaders {
+		ctx.Resp.Headers[key] = value
+	}
+}
+
 func findPods(ctx *types.SchedulingContext, names ...k8stypes.NamespacedName) []types.Pod {
 	res := []types.Pod{}
 	for _, pod := range ctx.PodsSnapshot {
diff --git a/pkg/epp/util/env/env_test.go b/pkg/epp/util/env/env_test.go
index 105beb2..f83104c 100644
--- a/pkg/epp/util/env/env_test.go
+++ b/pkg/epp/util/env/env_test.go
@@ -3,6 +3,7 @@ package env
 import (
 	"os"
 	"testing"
+	"time"
 
 	"github.com/go-logr/logr/testr"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
@@ -69,6 +70,80 @@ func TestGetEnvFloat(t *testing.T) {
 	}
 }
 
+func TestGetEnvDuration(t *testing.T) {
+	logger := testr.New(t)
+
+	tests := []struct {
+		name       string
+		key        string
+		value      string
+		defaultVal time.Duration
+		expected   time.Duration
+		setup      func()
+		teardown   func()
+	}{
+		{
+			name:       "env variable exists and is valid",
+			key:        "TEST_DURATION",
+			value:      "1h30m",
+			defaultVal: 0,
+			expected:   1*time.Hour + 30*time.Minute,
+			setup: func() {
+				os.Setenv("TEST_DURATION", "1h30m")
+			},
+			teardown: func() {
+				os.Unsetenv("TEST_DURATION")
+			},
+		},
+		{
+			name:       "env variable exists but is invalid",
+			key:        "TEST_DURATION",
+			value:      "invalid-duration",
+			defaultVal: 5 * time.Minute,
+			expected:   5 * time.Minute,
+			setup: func() {
+				os.Setenv("TEST_DURATION", "invalid-duration")
+			},
+			teardown: func() {
+				os.Unsetenv("TEST_DURATION")
+			},
+		},
+		{
+			name:       "env variable does not exist",
+			key:        "TEST_DURATION_MISSING",
+			defaultVal: 10 * time.Second,
+			expected:   10 * time.Second,
+			setup:      func() {},
+			teardown:   func() {},
+		},
+		{
+			name:       "env variable is empty string",
+			key:        "TEST_DURATION_EMPTY",
+			value:      "",
+			defaultVal: 1 * time.Millisecond,
+			expected:   1 * time.Millisecond,
+			setup: func() {
+				os.Setenv("TEST_DURATION_EMPTY", "")
+			},
+			teardown: func() {
+				os.Unsetenv("TEST_DURATION_EMPTY")
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			tc.setup()
+			defer tc.teardown()
+
+			result := GetEnvDuration(tc.key, tc.defaultVal, logger.V(logutil.VERBOSE))
+			if result != tc.expected {
+				t.Errorf("GetEnvDuration(%s, %v) = %v, expected %v", tc.key, tc.defaultVal, result, tc.expected)
+			}
+		})
+	}
+}
+
 func TestGetEnvInt(t *testing.T) {
 	logger := testr.New(t)
 
diff --git a/test/e2e/epp/e2e_suite_test.go b/test/e2e/epp/e2e_suite_test.go
index 01ed639..6595923 100644
--- a/test/e2e/epp/e2e_suite_test.go
+++ b/test/e2e/epp/e2e_suite_test.go
@@ -62,6 +62,8 @@ const (
 	modelServerName = "vllm-llama3-8b-instruct"
 	// modelName is the test model name.
 	modelName = "food-review"
+	// targetModelName is the target model name of the test model server.
+	targetModelName = modelName + "-1"
 	// envoyName is the name of the envoy proxy test resources.
 	envoyName = "envoy"
 	// envoyPort is the listener port number of the test envoy proxy.
diff --git a/test/e2e/epp/e2e_test.go b/test/e2e/epp/e2e_test.go
index 7240ceb..6fce63a 100644
--- a/test/e2e/epp/e2e_test.go
+++ b/test/e2e/epp/e2e_test.go
@@ -96,12 +96,8 @@ var _ = ginkgo.Describe("InferencePool", func() {
 func newInferenceModel(ns string) *v1alpha2.InferenceModel {
 	targets := []v1alpha2.TargetModel{
 		{
-			Name:   modelName,
-			Weight: ptr.To(int32(50)),
-		},
-		{
-			Name:   "cad-fabricator",
-			Weight: ptr.To(int32(50)),
+			Name:   targetModelName,
+			Weight: ptr.To(int32(100)),
 		},
 	}
 	return testutils.MakeModelWrapper("inferencemodel-sample", ns).
diff --git a/test/integration/epp/hermetic_test.go b/test/integration/epp/hermetic_test.go
index 68787da..9387876 100644
--- a/test/integration/epp/hermetic_test.go
+++ b/test/integration/epp/hermetic_test.go
@@ -155,6 +155,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte(strconv.Itoa(76)),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "hi",
+												RawValue: []byte("mom"),
+											},
+										},
 									}},
 							},
 						},
@@ -239,6 +245,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte(strconv.Itoa(76)),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "hi",
+												RawValue: []byte("mom"),
+											},
+										},
 									}},
 							},
 						},
@@ -323,6 +335,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte(strconv.Itoa(76)),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "hi",
+												RawValue: []byte("mom"),
+											},
+										},
 									}},
 							},
 						},
@@ -456,6 +474,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte(strconv.Itoa(76)),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "hi",
+												RawValue: []byte("mom"),
+											},
+										},
 									}},
 							},
 						},
@@ -567,6 +591,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte(strconv.Itoa(76)),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "hi",
+												RawValue: []byte("mom"),
+											},
+										},
 									}},
 							},
 						},
@@ -678,6 +708,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte(strconv.Itoa(74)),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "hi",
+												RawValue: []byte("mom"),
+											},
+										},
 									}},
 							},
 						},
@@ -778,6 +814,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte("true"),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "content-type",
+												RawValue: []uint8("application/json"),
+											},
+										},
 									},
 								},
 							},
@@ -877,6 +919,12 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte("true"),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "content-type",
+												RawValue: []uint8("application/json"),
+											},
+										},
 									},
 								},
 							},
@@ -1014,6 +1062,18 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 												RawValue: []byte("true"),
 											},
 										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "content-type",
+												RawValue: []byte("text/event-stream"),
+											},
+										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "status",
+												RawValue: []byte("200"),
+											},
+										},
 									},
 								},
 							},
diff --git a/test/testdata/inferencepool-e2e.yaml b/test/testdata/inferencepool-e2e.yaml
index 41bc1e1..0cd8934 100644
--- a/test/testdata/inferencepool-e2e.yaml
+++ b/test/testdata/inferencepool-e2e.yaml
@@ -73,13 +73,13 @@ spec:
         livenessProbe:
           grpc:
             port: 9003
-            service: envoy.service.ext_proc.v3.ExternalProcessor
+            service: inference-extension
           initialDelaySeconds: 5
           periodSeconds: 10
         readinessProbe:
           grpc:
             port: 9003
-            service: envoy.service.ext_proc.v3.ExternalProcessor
+            service: inference-extension
           initialDelaySeconds: 5
           periodSeconds: 10
 ---