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

Issue:
Add a generic AddPlugin option to configure the scheduler
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:

An `AddPlugin` implemented as a `type ConfigOption func(*SchedulerConfig)` that uses reflection to check which extension points the plugin implements, and add those extension points to the config.

This should also deprecate any existing helper functions to add plugins, such as `AddPrefixPlugin`

**Why is this needed**:

This is a follow up to the discussion in https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/768#pullrequestreview-2827459852

Discussion:
@liu-cong: /assign
@nirrozenbaum: /assign
(coordinated with @liu-cong)

Solution patch:
diff --git a/README.md b/README.md
index 9a729c1..87ee414 100644
--- a/README.md
+++ b/README.md
@@ -7,9 +7,11 @@
 Gateway API Inference Extension optimizes self-hosting Generative Models on Kubernetes.
 This is achieved by leveraging Envoy's [External Processing] (ext-proc) to extend any gateway that supports both ext-proc and [Gateway API] into an **[inference gateway]**. 
 
-
 [Inference Gateway]:#concepts-and-definitions
 
+## New!
+Inference Gateway has partnered with vLLM to accelerate LLM serving optimizations with [llm-d](https://llm-d.ai/blog/llm-d-announce)!
+
 ## Concepts and Definitions
 
 The following specific terms to this project:
@@ -53,6 +55,8 @@ For deeper insights and more advanced concepts, refer to our [proposals](/docs/p
 [LoRA Adapters]:https://docs.vllm.ai/en/stable/features/lora.html
 [External Processing]:https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_filters/ext_proc_filter
 
+
+
 ## Technical Overview
 
 This extension upgrades an [ext-proc](https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_filters/ext_proc_filter) capable proxy or gateway - such as Envoy Gateway, kGateway, or the GKE Gateway - to become an **[inference gateway]** - supporting inference platform teams self-hosting Generative Models (with a current focus on large language models) on Kubernetes. This integration makes it easy to expose and control access to your local [OpenAI-compatible chat completion endpoints](https://platform.openai.com/docs/api-reference/chat) to other workloads on or off cluster, or to integrate your self-hosted models alongside model-as-a-service providers in a higher level **AI Gateway** like LiteLLM, Solo AI Gateway, or Apigee.
@@ -66,7 +70,14 @@ The Inference Gateway:
 
 ![Architecture Diagram](./docs/inference-gateway-architecture.svg)
 
-It currently requires a version of vLLM that supports the necessary metrics to predict traffic load which is defined in the [model server protocol](https://github.com/kubernetes-sigs/gateway-api-inference-extension/tree/main/docs/proposals/003-model-server-protocol).  Support for Google's Jetstream, nVidia Triton, text-generation-inference, and SGLang is coming soon.
+### Model Server Integration
+
+IGW’s pluggable architecture was leveraged to enable the [llm-d Inference Scheduler](https://github.com/llm-d/llm-d-inference-scheduler).  
+
+Llm-d customizes vLLM & IGW to create a disaggregated serving solution. We've worked closely with this team to enable this integration. IGW will continue to work closely with llm-d to generalize the disaggregated serving plugin(s), & set a standard for disaggregated serving to be used across any [protocol-adherent](https://github.com/kubernetes-sigs/gateway-api-inference-extension/tree/main/docs/proposals/003-model-server-protocol) model server. 
+
+IGW has enhanced support for vLLM via llm-d, and broad support for any model servers implementing the protocol. More details can be found in [model server integration](https://gateway-api-inference-extension.sigs.k8s.io/implementations/model-servers/).
+
 
 ## Status
 
diff --git a/cmd/epp/main.go b/cmd/epp/main.go
index 9c023f2..44ad9f0 100644
--- a/cmd/epp/main.go
+++ b/cmd/epp/main.go
@@ -44,7 +44,6 @@ 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/multi/prefix"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/picker"
@@ -196,23 +195,21 @@ func run() error {
 	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{}
+
+		schedulerConfig := scheduling.NewSchedulerConfig().
+			WithFilters(filter.NewSheddableCapacityFilter()).
+			WithScorers(scorer.NewWeightedScorer(&scorer.QueueScorer{}, queueScorerWeight),
+				scorer.NewWeightedScorer(&scorer.KVCacheScorer{}, kvCacheScorerWeight)).
+			WithPicker(picker.NewMaxScorePicker())
+
 		if prefixCacheScheduling == "true" {
 			prefixScorerWeight := envutil.GetEnvInt("PREFIX_CACHE_SCORE_WEIGHT", prefix.DefaultScorerWeight, setupLog)
-			schedConfigOpts = append(schedConfigOpts, scheduling.AddPrefixPlugin(loadPrefixCacheConfig(), prefixScorerWeight))
+			if err := schedulerConfig.AddPlugins(scorer.NewWeightedScorer(prefix.New(loadPrefixCacheConfig()), prefixScorerWeight)); err != nil {
+				setupLog.Error(err, "Failed to register scheduler plugins")
+				return err
+			}
 		}
-		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{
diff --git a/go.mod b/go.mod
index 9888e7b..93ee8b9 100644
--- a/go.mod
+++ b/go.mod
@@ -3,26 +3,28 @@ module sigs.k8s.io/gateway-api-inference-extension
 go 1.24.0
 
 require (
+	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/elastic/crd-ref-docs v0.1.0
 	github.com/envoyproxy/go-control-plane/envoy v1.32.4
 	github.com/go-logr/logr v1.4.2
 	github.com/google/go-cmp v0.7.0
+	github.com/google/uuid v1.6.0
 	github.com/onsi/ginkgo/v2 v2.23.4
 	github.com/onsi/gomega v1.37.0
 	github.com/prometheus/client_golang v1.22.0
 	github.com/prometheus/client_model v0.6.2
-	github.com/prometheus/common v0.63.0
+	github.com/prometheus/common v0.64.0
 	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.71.1
 	google.golang.org/protobuf v1.36.6
-	k8s.io/api v0.32.4
-	k8s.io/apiextensions-apiserver v0.32.4
-	k8s.io/apimachinery v0.32.4
-	k8s.io/client-go v0.32.4
-	k8s.io/code-generator v0.32.4
-	k8s.io/component-base v0.32.4
+	k8s.io/api v0.32.5
+	k8s.io/apiextensions-apiserver v0.32.5
+	k8s.io/apimachinery v0.32.5
+	k8s.io/client-go v0.32.5
+	k8s.io/code-generator v0.32.5
+	k8s.io/component-base v0.32.5
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
 	sigs.k8s.io/controller-runtime v0.20.4
 	sigs.k8s.io/gateway-api v1.3.0
@@ -40,7 +42,6 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	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-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
@@ -67,7 +68,6 @@ require (
 	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.5.1 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
 	github.com/huandu/xstrings v1.3.3 // indirect
@@ -105,15 +105,15 @@ require (
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/automaxprocs v1.6.0 // indirect
-	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/crypto v0.38.0 // indirect
 	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.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
-	golang.org/x/text v0.24.0 // indirect
+	golang.org/x/net v0.40.0 // indirect
+	golang.org/x/oauth2 v0.30.0 // indirect
+	golang.org/x/sync v0.14.0 // indirect
+	golang.org/x/sys v0.33.0 // indirect
+	golang.org/x/term v0.32.0 // indirect
+	golang.org/x/text v0.25.0 // indirect
 	golang.org/x/time v0.7.0 // indirect
 	golang.org/x/tools v0.31.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
@@ -124,7 +124,7 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiserver v0.32.4 // indirect
+	k8s.io/apiserver v0.32.5 // indirect
 	k8s.io/gengo/v2 v2.0.0-20240911193312-2b36238f13e9 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f // indirect
diff --git a/go.sum b/go.sum
index 508f9ea..007ce14 100644
--- a/go.sum
+++ b/go.sum
@@ -162,8 +162,8 @@ github.com/prometheus/client_golang v1.22.0 h1:rb93p9lokFEsctTys46VnV1kLCDpVZ0a/
 github.com/prometheus/client_golang v1.22.0/go.mod h1:R7ljNsLXhuQXYZYtw6GAE9AZg8Y7vEW5scdCXrWRXC0=
 github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
 github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
-github.com/prometheus/common v0.63.0 h1:YR/EIY1o3mEFP/kZCD7iDMnLPlGyuU2Gb3HIcXnA98k=
-github.com/prometheus/common v0.63.0/go.mod h1:VVFF/fBIoToEnWRVkYoXEkq3R3paCoxG9PXP74SnV18=
+github.com/prometheus/common v0.64.0 h1:pdZeA+g617P7oGv1CzdTzyeShxAGrTBsolKNOLQPGO4=
+github.com/prometheus/common v0.64.0/go.mod h1:0gZns+BLRQ3V6NdaerOhMbwwRbNh9hkGINtQAsP5GS8=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
@@ -220,8 +220,8 @@ go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
-golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
+golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8=
+golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw=
 golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 h1:2dVuKD2vS7b0QIHQbpyTISPd0LeHDbnYEryqj5Q1ug8=
 golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56/go.mod h1:M4RDyNAINzryxdtnbRXRL/OHtkFuWGRjvuhBJpk2IlY=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -232,29 +232,29 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 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.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
-golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/net v0.40.0 h1:79Xs7wF06Gbdcg4kdCCIQArK11Z1hr5POQ6+fIYHNuY=
+golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds=
+golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI=
+golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU=
 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=
-golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
-golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
+golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
-golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
-golang.org/x/term v0.31.0 h1:erwDkOK1Msy6offm1mOgvspSkslFnIGsFnxOKoufg3o=
-golang.org/x/term v0.31.0/go.mod h1:R4BeIy7D95HzImkxGkTW1UQTtP54tio2RyHz7PwK0aw=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/term v0.32.0 h1:DR4lr0TjUs3epypdhTOkMmuF5CDFJ/8pOnbzMZPQ7bg=
+golang.org/x/term v0.32.0/go.mod h1:uZG1FhGx848Sqfsq4/DlJr3xGGsYMu/L5GW4abiaEPQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
-golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
+golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
+golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
 golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ=
 golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -293,20 +293,20 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.32.4 h1:kw8Y/G8E7EpNy7gjB8gJZl3KJkNz8HM2YHrZPtAZsF4=
-k8s.io/api v0.32.4/go.mod h1:5MYFvLvweRhyKylM3Es/6uh/5hGp0dg82vP34KifX4g=
-k8s.io/apiextensions-apiserver v0.32.4 h1:IA+CoR63UDOijR/vEpow6wQnX4V6iVpzazJBskHrpHE=
-k8s.io/apiextensions-apiserver v0.32.4/go.mod h1:Y06XO/b92H8ymOdG1HlA1submf7gIhbEDc3RjriqZOs=
-k8s.io/apimachinery v0.32.4 h1:8EEksaxA7nd7xWJkkwLDN4SvWS5ot9g6Z/VZb3ju25I=
-k8s.io/apimachinery v0.32.4/go.mod h1:GpHVgxoKlTxClKcteaeuF1Ul/lDVb74KpZcxcmLDElE=
-k8s.io/apiserver v0.32.4 h1:Yf7sd/y+GOQKH1Qf6wUeayZrYXe2SKZ17Bcq7VQM5HQ=
-k8s.io/apiserver v0.32.4/go.mod h1:JFUMNtE2M5yqLZpIsgCb06SkVSW1YcxW1oyLSTfjXR8=
-k8s.io/client-go v0.32.4 h1:zaGJS7xoYOYumoWIFXlcVrsiYioRPrXGO7dBfVC5R6M=
-k8s.io/client-go v0.32.4/go.mod h1:k0jftcyYnEtwlFW92xC7MTtFv5BNcZBr+zn9jPlT9Ic=
-k8s.io/code-generator v0.32.4 h1:d4dm/43RD6xhPBX22JgJw9JUpwTKzVR6tAxJD7pz83o=
-k8s.io/code-generator v0.32.4/go.mod h1:R0bKdIg1smtvsKvj9q7SxTeKq5X9ko6PuICCGt4yqxg=
-k8s.io/component-base v0.32.4 h1:HuF+2JVLbFS5GODLIfPCb1Td6b+G2HszJoArcWOSr5I=
-k8s.io/component-base v0.32.4/go.mod h1:10KloJEYw1keU/Xmjfy9TKJqUq7J2mYdiD1VDXoco4o=
+k8s.io/api v0.32.5 h1:uqjjsYo1kTJr5NIcoIaP9F+TgXgADH7nKQx91FDAhtk=
+k8s.io/api v0.32.5/go.mod h1:bXXFU3fGCZ/eFMZvfHZC69PeGbXEL4zzjuPVzOxHF64=
+k8s.io/apiextensions-apiserver v0.32.5 h1:o0aKvmzIIs8Uk54pidk32pxET+Pg2ULnh9WI1PuKTwE=
+k8s.io/apiextensions-apiserver v0.32.5/go.mod h1:5fpedJa3HJJFBukAZ6ur91DEDye5gYuXISPbOiNLYpU=
+k8s.io/apimachinery v0.32.5 h1:6We3aJ6crC0ap8EhsEXcgX3LpI6SEjubpiOMXLROwPM=
+k8s.io/apimachinery v0.32.5/go.mod h1:GpHVgxoKlTxClKcteaeuF1Ul/lDVb74KpZcxcmLDElE=
+k8s.io/apiserver v0.32.5 h1:phmm2EOUVFI+cLiq8Grtuh166fTt/qgvkGPkpgzp5uY=
+k8s.io/apiserver v0.32.5/go.mod h1:5bfueS1tgARVWVXRJBMI5mHoCmev0jOvbxebai/kiqc=
+k8s.io/client-go v0.32.5 h1:huFmQMzgWu0z4kbWsuZci+Gt4Fo72I4CcrvhToZ/Qp0=
+k8s.io/client-go v0.32.5/go.mod h1:Qchw6f9WIVrur7DKojAHpRgGLcANT0RLIvF39Jz58xA=
+k8s.io/code-generator v0.32.5 h1:dvoXgaWTDPLsg0txUzWj5xPV8UwHOsBhmm4JC9Gd1Qo=
+k8s.io/code-generator v0.32.5/go.mod h1:7S6jUv4ZAnI2yDUJUQUEuc3gv6+qFhnkB5Fhs9Eb0d8=
+k8s.io/component-base v0.32.5 h1:2HiX+m3s9Iz5CMqdCVDH2V942UqzQvjuhcXb4W+KCsg=
+k8s.io/component-base v0.32.5/go.mod h1:jDsPNFFElv9m27TcYxlpEX7TZ3vdgx2g4PaqMUHpV/Y=
 k8s.io/gengo/v2 v2.0.0-20240911193312-2b36238f13e9 h1:si3PfKm8dDYxgfbeA6orqrtLkvvIeH8UqffFJDl0bz4=
 k8s.io/gengo/v2 v2.0.0-20240911193312-2b36238f13e9/go.mod h1:EJykeLsmFC60UQbYJezXkEsG2FLrt0GPNkU5iK5GWxU=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go
index be85ba6..4b849c8 100644
--- a/pkg/epp/handlers/server.go
+++ b/pkg/epp/handlers/server.go
@@ -37,6 +37,11 @@ import (
 	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
 )
 
+const (
+	// Certain envoy implementations set a max limit of 64Kb per streamed chunk, intentionally setting this lower for a safe margin.
+	bodyByteLimit = 62000
+)
+
 func NewStreamingServer(destinationEndpointHintMetadataNamespace, destinationEndpointHintKey string, datastore Datastore, director Director) *StreamingServer {
 	return &StreamingServer{
 		destinationEndpointHintMetadataNamespace: destinationEndpointHintMetadataNamespace,
@@ -196,8 +201,8 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				err = json.Unmarshal(body, &reqCtx.Request.Body)
 				if err != nil {
 					logger.V(logutil.DEFAULT).Error(err, "Error unmarshaling request body")
-					// TODO: short circuit and send the body back as is (this could be an envoy error), currently we drop
-					// whatever the body request would have been and send our immediate response instead.
+					err = errutil.Error{Code: errutil.BadRequest, Msg: "Error unmarshaling request body: " + string(body)}
+					break
 				}
 
 				// Body stream complete. Allocate empty slice for response to use.
@@ -287,7 +292,24 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 					var responseErr error
 					responseErr = json.Unmarshal(body, &responseBody)
 					if responseErr != nil {
-						logger.V(logutil.DEFAULT).Error(responseErr, "Error unmarshaling request body")
+						logger.V(logutil.DEFAULT).Error(responseErr, "Error unmarshaling request body", "body", string(body))
+						reqCtx.respBodyResp = &extProcPb.ProcessingResponse{
+							Response: &extProcPb.ProcessingResponse_ResponseBody{
+								ResponseBody: &extProcPb.BodyResponse{
+									Response: &extProcPb.CommonResponse{
+										BodyMutation: &extProcPb.BodyMutation{
+											Mutation: &extProcPb.BodyMutation_StreamedResponse{
+												StreamedResponse: &extProcPb.StreamedBodyResponse{
+													Body:        body,
+													EndOfStream: true,
+												},
+											},
+										},
+									},
+								},
+							},
+						}
+						break
 					}
 
 					reqCtx, responseErr = s.HandleResponseBody(ctx, reqCtx, responseBody)
@@ -436,5 +458,39 @@ func BuildErrResponse(err error) (*extProcPb.ProcessingResponse, error) {
 	default:
 		return nil, status.Errorf(status.Code(err), "failed to handle request: %v", err)
 	}
+
+	if err.Error() != "" {
+		resp.Response.(*extProcPb.ProcessingResponse_ImmediateResponse).ImmediateResponse.Body = []byte(err.Error())
+	}
+
 	return resp, nil
 }
+
+func buildCommonResponses(bodyBytes []byte, byteLimit int) []*extProcPb.CommonResponse {
+	responses := []*extProcPb.CommonResponse{}
+	startingIndex := 0
+	bodyLen := len(bodyBytes)
+
+	for startingIndex < bodyLen {
+		eos := false
+		len := min(bodyLen-startingIndex, byteLimit)
+		chunk := bodyBytes[startingIndex : len+startingIndex]
+		if len+startingIndex == bodyLen {
+			eos = true
+		}
+
+		commonResp := &extProcPb.CommonResponse{
+			BodyMutation: &extProcPb.BodyMutation{
+				Mutation: &extProcPb.BodyMutation_StreamedResponse{
+					StreamedResponse: &extProcPb.StreamedBodyResponse{
+						Body:        chunk,
+						EndOfStream: eos,
+					},
+				},
+			},
+		}
+		responses = append(responses, commonResp)
+		startingIndex += len
+	}
+	return responses
+}
diff --git a/pkg/epp/scheduling/config.go b/pkg/epp/scheduling/config.go
deleted file mode 100644
index 77fb26f..0000000
--- a/pkg/epp/scheduling/config.go
+++ /dev/null
@@ -1,62 +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 scheduling
-
-import (
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/multi/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, 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.
-type SchedulerConfig struct {
-	preSchedulePlugins  []plugins.PreSchedule
-	filters             []plugins.Filter
-	scorers             map[plugins.Scorer]int // map from scorer to weight
-	picker              plugins.Picker
-	postSchedulePlugins []plugins.PostSchedule
-	postResponsePlugins []plugins.PostResponse
-}
-
-type ConfigOption func(*SchedulerConfig)
-
-// 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
index 399780b..066a90d 100644
--- a/pkg/epp/scheduling/plugins/filter/decision_tree_filter.go
+++ b/pkg/epp/scheduling/plugins/filter/decision_tree_filter.go
@@ -22,6 +22,9 @@ import (
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
+// compile-time type validation
+var _ plugins.Filter = &DecisionTreeFilter{}
+
 // 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.
diff --git a/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go b/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go
index eed6476..ee7d824 100644
--- a/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go
+++ b/pkg/epp/scheduling/plugins/filter/least_kvcache_filter.go
@@ -26,7 +26,7 @@ import (
 // compile-time type validation
 var _ plugins.Filter = &LeastKVCacheFilter{}
 
-// NewLeastKVCacheFilter returns a new LeastKVCacheFilter.
+// NewLeastKVCacheFilter initializes a new LeastKVCacheFilter and returns its pointer.
 func NewLeastKVCacheFilter() *LeastKVCacheFilter {
 	return &LeastKVCacheFilter{}
 }
diff --git a/pkg/epp/scheduling/plugins/filter/least_queue_filter.go b/pkg/epp/scheduling/plugins/filter/least_queue_filter.go
index 7a71d51..7e0bfc3 100644
--- a/pkg/epp/scheduling/plugins/filter/least_queue_filter.go
+++ b/pkg/epp/scheduling/plugins/filter/least_queue_filter.go
@@ -26,7 +26,7 @@ import (
 // compile-time type validation
 var _ plugins.Filter = &LeastQueueFilter{}
 
-// NewLeastQueueFilter returns a new LeastQueueFilter.
+// NewLeastQueueFilter initializes a new LeastQueueFilter and returns its pointer.
 func NewLeastQueueFilter() *LeastQueueFilter {
 	return &LeastQueueFilter{}
 }
@@ -35,8 +35,7 @@ func NewLeastQueueFilter() *LeastQueueFilter {
 // (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.
+// 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.
diff --git a/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go b/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go
index bc744a8..bc1e55b 100644
--- a/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go
+++ b/pkg/epp/scheduling/plugins/filter/lora_affinity_filter.go
@@ -28,7 +28,7 @@ import (
 // compile-time type validation
 var _ plugins.Filter = &LoraAffinityFilter{}
 
-// NewLoraAffinityFilter returns a new LoraAffinityFilter.
+// NewLoraAffinityFilter initializes a new LoraAffinityFilter and returns its pointer.
 func NewLoraAffinityFilter() *LoraAffinityFilter {
 	return &LoraAffinityFilter{
 		loraAffinityThreshold: config.Conf.LoraAffinityThreshold,
diff --git a/pkg/epp/scheduling/plugins/filter/low_queue_filter.go b/pkg/epp/scheduling/plugins/filter/low_queue_filter.go
index feb599b..ce7d352 100644
--- a/pkg/epp/scheduling/plugins/filter/low_queue_filter.go
+++ b/pkg/epp/scheduling/plugins/filter/low_queue_filter.go
@@ -25,7 +25,7 @@ import (
 // compile-time type validation
 var _ plugins.Filter = &LowQueueFilter{}
 
-// NewLowQueueFilter returns a new LowQueueFilter.
+// NewLowQueueFilter initializes a new LowQueueFilter and returns its pointer.
 func NewLowQueueFilter() *LowQueueFilter {
 	return &LowQueueFilter{
 		queueingThresholdLoRA: config.Conf.QueueingThresholdLoRA,
diff --git a/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go b/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go
index 5a298a0..cdc3355 100644
--- a/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go
+++ b/pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go
@@ -25,7 +25,7 @@ import (
 // compile-time type validation
 var _ plugins.Filter = &SheddableCapacityFilter{}
 
-// NewSheddableCapacityFilter returns a new SheddableCapacityFilter.
+// NewSheddableCapacityFilter initializes a new SheddableCapacityFilter and returns its pointer.
 func NewSheddableCapacityFilter() *SheddableCapacityFilter {
 	return &SheddableCapacityFilter{
 		queueThreshold:   config.Conf.QueueThresholdCritical,
diff --git a/pkg/epp/scheduling/plugins/multi/prefix/plugin.go b/pkg/epp/scheduling/plugins/multi/prefix/plugin.go
index de2cf70..b07f085 100644
--- a/pkg/epp/scheduling/plugins/multi/prefix/plugin.go
+++ b/pkg/epp/scheduling/plugins/multi/prefix/plugin.go
@@ -23,6 +23,7 @@ import (
 	"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/plugins"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
@@ -87,6 +88,7 @@ func (s ServerID) String() string {
 	return k8stypes.NamespacedName(s).String()
 }
 
+// compile-time type validation
 var _ types.StateData = &schedulingContextState{}
 
 // This is the state of this plugin to be used during a scheduling cycle.
@@ -111,6 +113,12 @@ func (s *schedulingContextState) Clone() types.StateData {
 	}
 }
 
+// compile-time type validation
+var _ plugins.PreSchedule = &Plugin{}
+var _ plugins.Scorer = &Plugin{}
+var _ plugins.PostSchedule = &Plugin{}
+
+// New initializes a new prefix Plugin and returns its pointer.
 func New(config Config) *Plugin {
 	m := &Plugin{
 		Config:  config,
@@ -124,6 +132,7 @@ func (m *Plugin) Name() string {
 	return "prefix-cache"
 }
 
+// PreSchedule initializes the prefix plugin state for the current scheduling cycle.
 func (m *Plugin) PreSchedule(ctx *types.SchedulingContext) {
 	hashes := hashPrompt(ctx, m.HashBlockSize, m.MaxPrefixBlocksToMatch)
 	state := &schedulingContextState{
@@ -135,7 +144,7 @@ func (m *Plugin) PreSchedule(ctx *types.SchedulingContext) {
 	ctx.Logger.V(logutil.TRACE).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:
+// PostSchedule records in the plugin cache the result of the scheduling selection.
 func (m *Plugin) PostSchedule(ctx *types.SchedulingContext, res *types.Result) {
 	targetPod := res.TargetPod.GetPod()
 	state, err := m.getPrefixState(ctx.CycleState)
@@ -199,6 +208,7 @@ func (m *Plugin) matchLongestPrefix(ctx *types.SchedulingContext, hashes []Block
 	return res
 }
 
+// getPrefixState returns the cycle state as a schedulingContextState.
 func (m *Plugin) getPrefixState(cycleState *types.CycleState) (*schedulingContextState, error) {
 	prefixStateKey := types.StateKey(m.Name())
 	state, err := cycleState.Read(prefixStateKey)
diff --git a/pkg/epp/scheduling/plugins/picker/max_score_picker.go b/pkg/epp/scheduling/plugins/picker/max_score_picker.go
index a6d7b39..e85aee5 100644
--- a/pkg/epp/scheduling/plugins/picker/max_score_picker.go
+++ b/pkg/epp/scheduling/plugins/picker/max_score_picker.go
@@ -24,11 +24,13 @@ import (
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
+// compile-time type validation
 var _ plugins.Picker = &MaxScorePicker{}
 
-func NewMaxScorePicker() plugins.Picker {
+// NewMaxScorePicker initializes a new MaxScorePicker and returns its pointer.
+func NewMaxScorePicker() *MaxScorePicker {
 	return &MaxScorePicker{
-		random: &RandomPicker{},
+		random: NewRandomPicker(),
 	}
 }
 
diff --git a/pkg/epp/scheduling/plugins/picker/random_picker.go b/pkg/epp/scheduling/plugins/picker/random_picker.go
index fb9f9a2..1d12198 100644
--- a/pkg/epp/scheduling/plugins/picker/random_picker.go
+++ b/pkg/epp/scheduling/plugins/picker/random_picker.go
@@ -25,15 +25,23 @@ import (
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
+// compile-time type validation
 var _ plugins.Picker = &RandomPicker{}
 
+// NewRandomPicker initializes a new RandomPicker and returns its pointer.
+func NewRandomPicker() *RandomPicker {
+	return &RandomPicker{}
+}
+
 // RandomPicker picks a random pod from the list of candidates.
 type RandomPicker struct{}
 
+// Name returns the name of the picker.
 func (p *RandomPicker) Name() string {
 	return "random"
 }
 
+// Pick selects a random pod from the list of candidates.
 func (p *RandomPicker) Pick(ctx *types.SchedulingContext, scoredPods []*types.ScoredPod) *types.Result {
 	ctx.Logger.V(logutil.DEBUG).Info(fmt.Sprintf("Selecting a random pod from %d candidates: %+v", len(scoredPods), scoredPods))
 	i := rand.Intn(len(scoredPods))
diff --git a/pkg/epp/scheduling/plugins/scorer/kvcache.go b/pkg/epp/scheduling/plugins/scorer/kvcache.go
index dbb6079..de6f87b 100644
--- a/pkg/epp/scheduling/plugins/scorer/kvcache.go
+++ b/pkg/epp/scheduling/plugins/scorer/kvcache.go
@@ -17,6 +17,7 @@ limitations under the License.
 package scorer
 
 import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
@@ -24,13 +25,19 @@ const (
 	DefaultKVCacheScorerWeight = 1
 )
 
+// compile-time type validation
+var _ plugins.Scorer = &KVCacheScorer{}
+
+// KVCacheScorer scores list of candidate pods based on KV cache utilization.
 type KVCacheScorer struct{}
 
-func (ss *KVCacheScorer) Name() string {
+// Name returns the name of the scorer.
+func (s *KVCacheScorer) Name() string {
 	return "kv-cache"
 }
 
-func (ss *KVCacheScorer) Score(ctx *types.SchedulingContext, pods []types.Pod) map[types.Pod]float64 {
+// Score returns the scoring result for the given list of pods based on context.
+func (s *KVCacheScorer) Score(ctx *types.SchedulingContext, pods []types.Pod) map[types.Pod]float64 {
 	scores := make(map[types.Pod]float64, len(pods))
 	for _, pod := range pods {
 		scores[pod] = 1 - pod.GetMetrics().KVCacheUsagePercent
diff --git a/pkg/epp/scheduling/plugins/scorer/queue.go b/pkg/epp/scheduling/plugins/scorer/queue.go
index bbe6b69..ac87f0b 100644
--- a/pkg/epp/scheduling/plugins/scorer/queue.go
+++ b/pkg/epp/scheduling/plugins/scorer/queue.go
@@ -19,6 +19,7 @@ package scorer
 import (
 	"math"
 
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
@@ -26,13 +27,20 @@ const (
 	DefaultQueueScorerWeight = 1
 )
 
+// compile-time type validation
+var _ plugins.Scorer = &QueueScorer{}
+
+// QueueScorer scores list of candidate pods based on the pod's waiting queue size.
+// the less waiting queue size the pod has, the higher score it will get (since it's more available to serve new request).
 type QueueScorer struct{}
 
-func (q *QueueScorer) Name() string {
+// Name returns the name of the scorer.
+func (s *QueueScorer) Name() string {
 	return "queue"
 }
 
-func (q *QueueScorer) Score(ctx *types.SchedulingContext, pods []types.Pod) map[types.Pod]float64 {
+// Score returns the scoring result for the given list of pods based on context.
+func (s *QueueScorer) Score(ctx *types.SchedulingContext, pods []types.Pod) map[types.Pod]float64 {
 	minQueueSize := math.MaxInt
 	maxQueueSize := math.MinInt
 
diff --git a/pkg/epp/scheduling/plugins/scorer/weighted_scorer.go b/pkg/epp/scheduling/plugins/scorer/weighted_scorer.go
new file mode 100644
index 0000000..1e83b6c
--- /dev/null
+++ b/pkg/epp/scheduling/plugins/scorer/weighted_scorer.go
@@ -0,0 +1,40 @@
+/*
+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 scorer
+
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+)
+
+// NewWeightedScorer initializes a new WeightedScorer and returns its pointer.
+func NewWeightedScorer(scorer plugins.Scorer, weight int) *WeightedScorer {
+	return &WeightedScorer{
+		Scorer: scorer,
+		weight: weight,
+	}
+}
+
+// WeightedScorer is a struct that encapsulates a scorer with its weight.
+type WeightedScorer struct {
+	plugins.Scorer
+	weight int
+}
+
+// Weight returns the weight of the scorer.
+func (s *WeightedScorer) Weight() int {
+	return s.weight
+}
diff --git a/pkg/epp/scheduling/scheduler.go b/pkg/epp/scheduling/scheduler.go
index b484cde..87591b2 100644
--- a/pkg/epp/scheduling/scheduler.go
+++ b/pkg/epp/scheduling/scheduler.go
@@ -28,6 +28,7 @@ import (
 	"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/scorer"
 	"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"
@@ -64,13 +65,9 @@ func NewScheduler(datastore Datastore) *Scheduler {
 		},
 	}
 
-	defaultConfig := &SchedulerConfig{
-		preSchedulePlugins:  []plugins.PreSchedule{},
-		filters:             []plugins.Filter{filter.NewSheddableCapacityFilter(), lowLatencyFilter},
-		scorers:             map[plugins.Scorer]int{},
-		picker:              &picker.RandomPicker{},
-		postSchedulePlugins: []plugins.PostSchedule{},
-	}
+	defaultConfig := NewSchedulerConfig().
+		WithFilters(filter.NewSheddableCapacityFilter(), lowLatencyFilter).
+		WithPicker(&picker.RandomPicker{})
 
 	return NewSchedulerWithConfig(datastore, defaultConfig)
 }
@@ -92,7 +89,7 @@ type Scheduler struct {
 	datastore           Datastore
 	preSchedulePlugins  []plugins.PreSchedule
 	filters             []plugins.Filter
-	scorers             map[plugins.Scorer]int // map from scorer to its weight
+	scorers             []*scorer.WeightedScorer
 	picker              plugins.Picker
 	postSchedulePlugins []plugins.PostSchedule
 	postResponsePlugins []plugins.PostResponse
@@ -172,15 +169,15 @@ func (s *Scheduler) runScorerPlugins(ctx *types.SchedulingContext, pods []types.
 		weightedScorePerPod[pod] = float64(0) // initialize weighted score per pod with 0 value
 	}
 	// Iterate through each scorer in the chain and accumulate the weighted scores.
-	for scorer, weight := range s.scorers {
-		loggerDebug.Info("Running scorer", "scorer", scorer.Name())
+	for _, weightedScorer := range s.scorers {
+		loggerDebug.Info("Running scorer", "scorer", weightedScorer.Name())
 		before := time.Now()
-		scores := scorer.Score(ctx, pods)
-		metrics.RecordSchedulerPluginProcessingLatency(plugins.ScorerPluginType, scorer.Name(), time.Since(before))
+		scores := weightedScorer.Score(ctx, pods)
+		metrics.RecordSchedulerPluginProcessingLatency(plugins.ScorerPluginType, weightedScorer.Name(), time.Since(before))
 		for pod, score := range scores { // weight is relative to the sum of weights
-			weightedScorePerPod[pod] += score * float64(weight) // TODO normalize score before multiply with weight
+			weightedScorePerPod[pod] += score * float64(weightedScorer.Weight())
 		}
-		loggerDebug.Info("After running scorer", "scorer", scorer.Name())
+		loggerDebug.Info("After running scorer", "scorer", weightedScorer.Name())
 	}
 	loggerDebug.Info("After running scorer plugins")
 
diff --git a/pkg/epp/scheduling/scheduler_config.go b/pkg/epp/scheduling/scheduler_config.go
new file mode 100644
index 0000000..06a23f4
--- /dev/null
+++ b/pkg/epp/scheduling/scheduler_config.go
@@ -0,0 +1,123 @@
+/*
+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 scheduling
+
+import (
+	"fmt"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/scorer"
+)
+
+// NewSchedulerConfig creates a new SchedulerConfig object and returns its pointer.
+func NewSchedulerConfig() *SchedulerConfig {
+	return &SchedulerConfig{
+		preSchedulePlugins:  []plugins.PreSchedule{},
+		filters:             []plugins.Filter{},
+		scorers:             []*scorer.WeightedScorer{},
+		postSchedulePlugins: []plugins.PostSchedule{},
+		postResponsePlugins: []plugins.PostResponse{},
+		// picker remains nil since config doesn't support multiple pickers
+	}
+}
+
+// SchedulerConfig provides a configuration for the scheduler which influence routing decisions.
+type SchedulerConfig struct {
+	preSchedulePlugins  []plugins.PreSchedule
+	filters             []plugins.Filter
+	scorers             []*scorer.WeightedScorer
+	picker              plugins.Picker
+	postSchedulePlugins []plugins.PostSchedule
+	postResponsePlugins []plugins.PostResponse
+}
+
+// WithPreSchedulePlugins sets the given plugins as the PreSchedule plugins.
+// If the SchedulerConfig has PreSchedule plugins, this call replaces the existing plugins with the given ones.
+func (c *SchedulerConfig) WithPreSchedulePlugins(plugins ...plugins.PreSchedule) *SchedulerConfig {
+	c.preSchedulePlugins = plugins
+	return c
+}
+
+// WithFilters sets the given filter plugins as the Filter plugins.
+// if the SchedulerConfig has Filter plugins, this call replaces the existing plugins with the given ones.
+func (c *SchedulerConfig) WithFilters(filters ...plugins.Filter) *SchedulerConfig {
+	c.filters = filters
+	return c
+}
+
+// WithScorers sets the given scorer plugins as the Scorer plugins.
+// if the SchedulerConfig has Scorer plugins, this call replaces the existing plugins with the given ones.
+func (c *SchedulerConfig) WithScorers(scorers ...*scorer.WeightedScorer) *SchedulerConfig {
+	c.scorers = scorers
+	return c
+}
+
+// WithPicker sets the given picker plugins as the Picker plugin.
+// if the SchedulerConfig has Picker plugin, this call replaces the existing plugin with the given one.
+func (c *SchedulerConfig) WithPicker(picker plugins.Picker) *SchedulerConfig {
+	c.picker = picker
+	return c
+}
+
+// WithPostSchedulePlugins sets the given plugins as the PostSchedule plugins.
+// If the SchedulerConfig has PostSchedule plugins, this call replaces the existing plugins with the given ones.
+func (c *SchedulerConfig) WithPostSchedulePlugins(plugins ...plugins.PostSchedule) *SchedulerConfig {
+	c.postSchedulePlugins = plugins
+	return c
+}
+
+// WithPostResponsePlugins sets the given plugins as the PostResponse plugins.
+// If the SchedulerConfig has PostResponse plugins, this call replaces the existing plugins with the given ones.
+func (c *SchedulerConfig) WithPostResponsePlugins(plugins ...plugins.PostResponse) *SchedulerConfig {
+	c.postResponsePlugins = plugins
+	return c
+}
+
+// AddPlugins adds the given plugins to all scheduler plugins according to the interfaces each plugin implements.
+// A plugin may implement more than one scheduler plugin interface.
+// Special Case: In order to add a scorer, one must use the scorer.NewWeightedScorer function in order to provide a weight.
+// if a scorer implements more than one interface, supplying a WeightedScorer is sufficient. The function will take the internal
+// scorer object and register it to all interfaces it implements.
+func (c *SchedulerConfig) AddPlugins(pluginObjects ...plugins.Plugin) error {
+	for _, plugin := range pluginObjects {
+		if weightedScorer, ok := plugin.(*scorer.WeightedScorer); ok {
+			c.scorers = append(c.scorers, weightedScorer)
+			plugin = weightedScorer.Scorer // if we got WeightedScorer, unwrap the plugin
+		} else if scorer, ok := plugin.(plugins.Scorer); ok { // if we got a Scorer instead of WeightedScorer that's an error.
+			return fmt.Errorf("failed to register scorer '%s' without a weight. follow function documentation to register a scorer", scorer.Name())
+		}
+		if preSchedulePlugin, ok := plugin.(plugins.PreSchedule); ok {
+			c.preSchedulePlugins = append(c.preSchedulePlugins, preSchedulePlugin)
+		}
+		if filter, ok := plugin.(plugins.Filter); ok {
+			c.filters = append(c.filters, filter)
+		}
+		if picker, ok := plugin.(plugins.Picker); ok {
+			if c.picker != nil {
+				return fmt.Errorf("failed to set '%s' as picker, already have a registered picker plugin '%s'", picker.Name(), c.picker.Name())
+			}
+			c.picker = picker
+		}
+		if postSchedulePlugin, ok := plugin.(plugins.PostSchedule); ok {
+			c.postSchedulePlugins = append(c.postSchedulePlugins, postSchedulePlugin)
+		}
+		if postResponsePlugin, ok := plugin.(plugins.PostResponse); ok {
+			c.postResponsePlugins = append(c.postResponsePlugins, postResponsePlugin)
+		}
+	}
+	return nil
+}
diff --git a/pkg/epp/scheduling/types/scheduling_context.go b/pkg/epp/scheduling/types/scheduling_context.go
index 3762180..d89ad3d 100644
--- a/pkg/epp/scheduling/types/scheduling_context.go
+++ b/pkg/epp/scheduling/types/scheduling_context.go
@@ -24,7 +24,6 @@ import (
 )
 
 func NewSchedulingContext(ctx context.Context, req *LLMRequest, resp *LLMResponse, pods []Pod) *SchedulingContext {
-
 	logger := log.FromContext(ctx).WithValues("request", req)
 	return &SchedulingContext{
 		Context:      ctx,
diff --git a/pkg/epp/server/runserver.go b/pkg/epp/server/runserver.go
index 4b86208..69f4805 100644
--- a/pkg/epp/server/runserver.go
+++ b/pkg/epp/server/runserver.go
@@ -47,7 +47,6 @@ type ExtProcServerRunner struct {
 	Datastore                                datastore.Datastore
 	SecureServing                            bool
 	CertPath                                 string
-	UseStreaming                             bool
 	RefreshPrometheusMetricsInterval         time.Duration
 	Scheduler                                requestcontrol.Scheduler
 
diff --git a/site-src/index.md b/site-src/index.md
index e7050ce..b304366 100644
--- a/site-src/index.md
+++ b/site-src/index.md
@@ -46,7 +46,7 @@ this project.
 
 ### Endpoint Picker
 
-As part of this project, we've built the Endpoing Picker. A pluggable & extensible ext-proc deployment that implements [this architecture](https://github.com/kubernetes-sigs/gateway-api-inference-extension/tree/main/docs/proposals/0683-epp-architecture-proposal).
+As part of this project, we've built the Endpoint Picker. A pluggable & extensible ext-proc deployment that implements [this architecture](https://github.com/kubernetes-sigs/gateway-api-inference-extension/tree/main/docs/proposals/0683-epp-architecture-proposal).
 
 ### Model Server Frameworks
 



Test patch:
diff --git a/pkg/epp/handlers/server_test.go b/pkg/epp/handlers/server_test.go
new file mode 100644
index 0000000..cc99a51
--- /dev/null
+++ b/pkg/epp/handlers/server_test.go
@@ -0,0 +1,64 @@
+package handlers
+
+import (
+	"crypto/rand"
+	"testing"
+)
+
+func TestBuildCommonResponses(t *testing.T) {
+	tests := []struct {
+		name                 string
+		count                int
+		expectedMessageCount int
+	}{
+		{
+			name:                 "below limit",
+			count:                bodyByteLimit - 1000,
+			expectedMessageCount: 1,
+		},
+		{
+			name:                 "at limit",
+			count:                bodyByteLimit,
+			expectedMessageCount: 1,
+		},
+		{
+			name:                 "off by one error?",
+			count:                bodyByteLimit + 1,
+			expectedMessageCount: 2,
+		},
+		{
+			name:                 "above limit",
+			count:                bodyByteLimit + 1000,
+			expectedMessageCount: 2,
+		},
+		{
+			name:                 "above limit",
+			count:                (bodyByteLimit * 2) + 1000,
+			expectedMessageCount: 3,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			arr := generateBytes(test.count)
+			responses := buildCommonResponses(arr, bodyByteLimit)
+			for i, response := range responses {
+				eos := response.BodyMutation.GetStreamedResponse().GetEndOfStream()
+				if eos == true && i+1 != len(responses) {
+					t.Fatalf("EoS should not be set")
+				}
+				if eos == false && i+1 == len(responses) {
+					t.Fatalf("EoS should be set")
+				}
+			}
+			if len(responses) != test.expectedMessageCount {
+				t.Fatalf("Expected: %v, Got %v", test.expectedMessageCount, len(responses))
+			}
+		})
+	}
+}
+
+func generateBytes(count int) []byte {
+	arr := make([]byte, count)
+	_, _ = rand.Read(arr)
+	return arr
+}
diff --git a/pkg/epp/scheduling/scheduler_test.go b/pkg/epp/scheduling/scheduler_test.go
index 933122e..1c6e649 100644
--- a/pkg/epp/scheduling/scheduler_test.go
+++ b/pkg/epp/scheduling/scheduler_test.go
@@ -26,6 +26,7 @@ import (
 	"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
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/scorer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 )
 
@@ -276,9 +277,9 @@ func TestSchedulePlugins(t *testing.T) {
 			config: SchedulerConfig{
 				preSchedulePlugins: []plugins.PreSchedule{tp1, tp2},
 				filters:            []plugins.Filter{tp1, tp2},
-				scorers: map[plugins.Scorer]int{
-					tp1: 1,
-					tp2: 1,
+				scorers: []*scorer.WeightedScorer{
+					scorer.NewWeightedScorer(tp1, 1),
+					scorer.NewWeightedScorer(tp2, 1),
 				},
 				picker:              pickerPlugin,
 				postSchedulePlugins: []plugins.PostSchedule{tp1, tp2},
@@ -298,9 +299,9 @@ func TestSchedulePlugins(t *testing.T) {
 			config: SchedulerConfig{
 				preSchedulePlugins: []plugins.PreSchedule{tp1, tp2},
 				filters:            []plugins.Filter{tp1, tp2},
-				scorers: map[plugins.Scorer]int{
-					tp1: 60,
-					tp2: 40,
+				scorers: []*scorer.WeightedScorer{
+					scorer.NewWeightedScorer(tp1, 60),
+					scorer.NewWeightedScorer(tp2, 40),
 				},
 				picker:              pickerPlugin,
 				postSchedulePlugins: []plugins.PostSchedule{tp1, tp2},
@@ -320,9 +321,9 @@ func TestSchedulePlugins(t *testing.T) {
 			config: SchedulerConfig{
 				preSchedulePlugins: []plugins.PreSchedule{tp1, tp2},
 				filters:            []plugins.Filter{tp1, tp_filterAll},
-				scorers: map[plugins.Scorer]int{
-					tp1: 1,
-					tp2: 1,
+				scorers: []*scorer.WeightedScorer{
+					scorer.NewWeightedScorer(tp1, 1),
+					scorer.NewWeightedScorer(tp2, 1),
 				},
 				picker:              pickerPlugin,
 				postSchedulePlugins: []plugins.PostSchedule{tp1, tp2},
@@ -346,8 +347,8 @@ func TestSchedulePlugins(t *testing.T) {
 			for _, plugin := range test.config.filters {
 				plugin.(*TestPlugin).reset()
 			}
-			for plugin := range test.config.scorers {
-				plugin.(*TestPlugin).reset()
+			for _, plugin := range test.config.scorers {
+				plugin.Scorer.(*TestPlugin).reset()
 			}
 			test.config.picker.(*TestPlugin).reset()
 			for _, plugin := range test.config.postSchedulePlugins {
@@ -396,8 +397,8 @@ func TestSchedulePlugins(t *testing.T) {
 				}
 			}
 
-			for plugin := range test.config.scorers {
-				tp, _ := plugin.(*TestPlugin)
+			for _, plugin := range test.config.scorers {
+				tp, _ := plugin.Scorer.(*TestPlugin)
 				if tp.ScoreCallCount != 1 {
 					t.Errorf("Plugin %s Score() called %d times, expected 1", plugin.Name(), tp.ScoreCallCount)
 				}
diff --git a/test/integration/epp/hermetic_test.go b/test/integration/epp/hermetic_test.go
index 9233925..a9d54fa 100644
--- a/test/integration/epp/hermetic_test.go
+++ b/test/integration/epp/hermetic_test.go
@@ -55,7 +55,6 @@ import (
 	metricsutils "k8s.io/component-base/metrics/testutil"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 	k8sclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/config"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -66,7 +65,6 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 	epptestutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/testing"
@@ -185,6 +183,74 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "invalid json; return body",
+			requests: []*extProcPb.ProcessingRequest{
+				{
+					Request: &extProcPb.ProcessingRequest_RequestHeaders{
+						RequestHeaders: &extProcPb.HttpHeaders{
+							Headers: &configPb.HeaderMap{
+								Headers: []*configPb.HeaderValue{
+									{
+										Key:   "hi",
+										Value: "mom",
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Request: &extProcPb.ProcessingRequest_RequestBody{
+						RequestBody: &extProcPb.HttpBody{Body: []byte("no healthy upstream"), EndOfStream: true},
+					},
+				},
+			},
+			// pod-1 will be picked because it has relatively low queue size, with the requested
+			// model being active, and has low KV cache.
+			pods: map[*backend.Pod]*backendmetrics.MetricsState{
+				fakePod(0): {
+					WaitingQueueSize:    0,
+					KVCacheUsagePercent: 0.2,
+					ActiveModels: map[string]int{
+						"foo": 1,
+						"bar": 1,
+					},
+					WaitingModels: map[string]int{},
+				},
+				fakePod(1): {
+					WaitingQueueSize:    0,
+					KVCacheUsagePercent: 0.1,
+					ActiveModels: map[string]int{
+						"foo":            1,
+						"sql-lora-1fdg2": 1,
+					},
+					WaitingModels: map[string]int{},
+				},
+				fakePod(2): {
+					WaitingQueueSize:    10,
+					KVCacheUsagePercent: 0.2,
+					ActiveModels: map[string]int{
+						"foo": 1,
+						"bar": 1,
+					},
+					WaitingModels: map[string]int{},
+				},
+			},
+			wantErr: false,
+			wantResponses: []*extProcPb.ProcessingResponse{
+				{
+					Response: &extProcPb.ProcessingResponse_ImmediateResponse{
+						ImmediateResponse: &extProcPb.ImmediateResponse{
+							Status: &envoyTypePb.HttpStatus{
+								Code: envoyTypePb.StatusCode_BadRequest,
+							},
+							Body: []byte("inference gateway: BadRequest - Error unmarshaling request body: no healthy upstream"),
+						},
+					},
+				},
+			},
+		},
 		{
 			name:     "select active lora, low queue",
 			requests: integrationutils.GenerateStreamedRequestSet(logger, "test2", "sql-lora"),
@@ -409,6 +475,7 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 							Status: &envoyTypePb.HttpStatus{
 								Code: envoyTypePb.StatusCode_TooManyRequests,
 							},
+							Body: []byte("inference gateway: InferencePoolResourceExhausted - failed to find target pod: inference gateway: Internal - no pods available for the given request"),
 						},
 					},
 				},
@@ -844,6 +911,106 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "Response is invalid json; return body",
+			requests: []*extProcPb.ProcessingRequest{
+				{
+					Request: &extProcPb.ProcessingRequest_ResponseHeaders{
+						ResponseHeaders: &extProcPb.HttpHeaders{
+							Headers: &configPb.HeaderMap{
+								Headers: []*configPb.HeaderValue{
+									{
+										Key:   "content-type",
+										Value: "application/json",
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Request: &extProcPb.ProcessingRequest_ResponseBody{
+						ResponseBody: &extProcPb.HttpBody{Body: []byte("no healthy upstream"), EndOfStream: true},
+					},
+				},
+			},
+
+			//
+			// pod 0 will be picked as all other models are above threshold
+			pods: map[*backend.Pod]*backendmetrics.MetricsState{
+				fakePod(0): {
+					WaitingQueueSize:    4,
+					KVCacheUsagePercent: 0.2,
+					ActiveModels: map[string]int{
+						"foo":            1,
+						"bar":            1,
+						"sql-lora-1fdg3": 1,
+					},
+					WaitingModels: map[string]int{},
+				},
+				fakePod(1): {
+					WaitingQueueSize:    0,
+					KVCacheUsagePercent: 0.85,
+					ActiveModels: map[string]int{
+						"foo":            1,
+						"sql-lora-1fdg3": 1,
+					},
+					WaitingModels: map[string]int{},
+				},
+				fakePod(2): {
+					WaitingQueueSize:    10,
+					KVCacheUsagePercent: 0.9,
+					ActiveModels: map[string]int{
+						"foo":            1,
+						"sql-lora-1fdg3": 1,
+					},
+					WaitingModels: map[string]int{},
+				},
+			},
+			wantErr: false,
+			wantResponses: []*extProcPb.ProcessingResponse{
+				{
+					Response: &extProcPb.ProcessingResponse_ResponseHeaders{
+						ResponseHeaders: &extProcPb.HeadersResponse{
+							Response: &extProcPb.CommonResponse{
+								HeaderMutation: &extProcPb.HeaderMutation{
+									SetHeaders: []*configPb.HeaderValueOption{
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "x-went-into-resp-headers",
+												RawValue: []byte("true"),
+											},
+										},
+										{
+											Header: &configPb.HeaderValue{
+												Key:      "content-type",
+												RawValue: []uint8("application/json"),
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Response: &extProcPb.ProcessingResponse_ResponseBody{
+						ResponseBody: &extProcPb.BodyResponse{
+							Response: &extProcPb.CommonResponse{
+								BodyMutation: &extProcPb.BodyMutation{
+									Mutation: &extProcPb.BodyMutation_StreamedResponse{
+										StreamedResponse: &extProcPb.StreamedBodyResponse{
+											Body:        []byte("no healthy upstream"),
+											EndOfStream: true,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "responsebody sent over a single request, but empty body with EndOfStream in the second request(this is how envoy operates); content-type is json, buffer",
 			requests: []*extProcPb.ProcessingRequest{
@@ -1263,14 +1430,19 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			client, cleanup := setUpHermeticServer(t, test.pods, true)
+			client, cleanup := setUpHermeticServer(t, test.pods)
 			t.Cleanup(cleanup)
 			responses, err := integrationutils.StreamedRequest(t, client, test.requests, len(test.wantResponses))
 
 			if err != nil && !test.wantErr {
 				t.Errorf("Unexpected error, got: %v, want error: %v", err, test.wantErr)
 			}
-			if diff := cmp.Diff(test.wantResponses, responses, protocmp.Transform()); diff != "" {
+			if diff := cmp.Diff(test.wantResponses, responses,
+				protocmp.Transform(),
+				protocmp.SortRepeated(func(a, b *configPb.HeaderValueOption) bool {
+					return a.GetHeader().GetKey() < b.GetHeader().GetKey()
+				}),
+			); diff != "" {
 				t.Errorf("Unexpected response, (-want +got): %v", diff)
 			}
 
@@ -1287,14 +1459,13 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 	}
 }
 
-func setUpHermeticServer(t *testing.T, podAndMetrics map[*backend.Pod]*backendmetrics.MetricsState, streamed bool) (client extProcPb.ExternalProcessor_ProcessClient, cleanup func()) {
+func setUpHermeticServer(t *testing.T, podAndMetrics map[*backend.Pod]*backendmetrics.MetricsState) (client extProcPb.ExternalProcessor_ProcessClient, cleanup func()) {
 	// Reconfigure the TestPodMetricsClient.
 	res := map[types.NamespacedName]*backendmetrics.MetricsState{}
 	for pod, metrics := range podAndMetrics {
 		res[pod.NamespacedName] = metrics
 	}
 	serverRunner.TestPodMetricsClient.SetRes(res)
-	serverRunner.UseStreaming = streamed
 
 	serverCtx, stopServer := context.WithCancel(context.Background())
 
@@ -1398,7 +1569,7 @@ func BeforeSuite() func() {
 	// Init runtime.
 	ctrl.SetLogger(logger)
 
-	mgr, err := server.NewManagerWithOptions(cfg, managerTestOptions("default", "vllm-llama3-8b-instruct-pool"))
+	mgr, err := runserver.NewManagerWithOptions(cfg, managerTestOptions("default", "vllm-llama3-8b-instruct-pool"))
 	if err != nil {
 		logutil.Fatal(logger, err, "Failed to create controller manager")
 	}
@@ -1537,7 +1708,7 @@ func managerTestOptions(namespace, name string) ctrl.Options {
 	return ctrl.Options{
 		Scheme: scheme,
 		Cache: cache.Options{
-			ByObject: map[client.Object]cache.ByObject{
+			ByObject: map[k8sclient.Object]cache.ByObject{
 				&corev1.Pod{}: {
 					Namespaces: map[string]cache.Config{
 						namespace: {},