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

Issue:
[Flow Control] Implement Dynamic Priority Band Configuration
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**: The Flow Control layer should dynamically configure its recognized priority bands based on the `InferenceObjective` resources registered to the pool. This will remove the current need for a static, at-startup configuration of the priority domain.

**Why is this needed**: Currently, only priority 0 is configured by default, and any other priority (including negative values for load shedding) must be hardcoded. This change will allow users to define and use any priority value in their `InferenceObjective` resources without needing to fork, modify, and redeploy the EPP. This is a significant usability improvement and makes the priority system far more flexible.

Discussion:
@LukeAVanDrie: /assign@LukeAVanDrie: Here is a quick summary of the problem shape.

*   **The Hot Path (Stats Propagation):**
    *   **Frequency:** Executes on every single Request Enqueue and Dequeue.
    *   **Concurrency:** High (called by every active request goroutine).
    *   **Constraint:** Must not block.
    *   **Characteristics:** Read-Only 99.999% of the time.

*   **The Cold Path (Dynamic Provisioning):**
    *   **Frequency:** Extremely Rare. Happens once per unique Priority Level process-lifetime (e.g., we have 10 priority bands, this runs 10 times total).
    *   **Concurrency:** Low.
    *   **Constraint:** Must be thread-safe (prevent double-creation). Latency is irrelevant.
    *   **Characteristics:** Write-Heavy (modifies the topology map).

---

| | `sync.RWMutex` | `sync.Map` | `atomic.Pointer` (COW) |
| :--- | :--- | :--- | :--- |
| **Code Style** | Idiomatic Go.| Standard Lib. Common for caches, but requires casting. | Custom implementation of lock-free logic. |
| **Type Safety** | Strong. Compiler-enforced. | Weak. Stores `any`. Requires runtime casting (`val.(*band)`). | Strong. Uses Go Generics (`atomic.Pointer[T]`). |
| **Hot Path Cost** | Updates a shared atomic counter. | Interface boxing + internal atomic checks. | Single pointer dereference. |
| **Maintenance** | Hard to break. | Risk of panic if types drift in future refactors, but practically mitigated by unit testing. | High. Easy to break the "Copy" logic in the write path. |
@LukeAVanDrie: I am leaning toward `sync.Map`. Will send out a draft PR with this approach.@LukeAVanDrie: To start, any un-configured band will get the default config. Later we can consider if we want to match the config of the nearest configured floor or ceiling. (e.g., if 10, 20, and 30 are configured and 31 comes in, use band 30 config).

Solution patch:
diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go
index b2fb299..444544a 100644
--- a/cmd/epp/runner/runner.go
+++ b/cmd/epp/runner/runner.go
@@ -94,16 +94,22 @@ const (
 	EnvSdMetricsStalenessThreshold = "SD_METRICS_STALENESS_THRESHOLD"
 )
 
-// TODO: this is hardcoded for POC only. This needs to be hooked up to our text-based config story.
+// TODO: This is a bad pattern. Remove this once we hook Flow Control into the config loading path.
+func must[T any](t T, err error) T {
+	if err != nil {
+		panic(fmt.Sprintf("static initialization failed: %v", err))
+	}
+	return t
+}
+
+// TODO: This is hardcoded for POC only. This needs to be hooked up to our text-based config story.
 var flowControlConfig = flowcontrol.Config{
 	Controller: fccontroller.Config{}, // Use all defaults.
-	Registry: fcregistry.Config{
-		// Define domain of accepted priority levels as this field is required. Use defaults for all optional fields.
-		// TODO: this should not be hardcoded.
-		PriorityBands: []fcregistry.PriorityBandConfig{
-			{Priority: 0, PriorityName: "Default"},
-		},
-	},
+	Registry: must(fcregistry.NewConfig(
+		fcregistry.WithPriorityBand(
+			must(fcregistry.NewPriorityBandConfig(0, "Default")),
+		),
+	)),
 }
 
 var (
diff --git a/go.mod b/go.mod
index 31309e6..78d0600 100644
--- a/go.mod
+++ b/go.mod
@@ -19,7 +19,7 @@ require (
 	github.com/prometheus/prometheus v0.307.3
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/otel v1.39.0
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.39.0
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.39.0
 	go.opentelemetry.io/otel/sdk v1.39.0
 	go.uber.org/multierr v1.11.0
@@ -79,7 +79,7 @@ require (
 	github.com/google/pprof v0.0.0-20250923004556-9e5a51aed1e8 // indirect
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
 	github.com/grafana/regexp v0.0.0-20250905093917-f7b3be9d1853 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 // indirect
 	github.com/huandu/xstrings v1.3.3 // indirect
 	github.com/imdario/mergo v0.3.16 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
@@ -102,26 +102,26 @@ require (
 	github.com/x448/float16 v0.8.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect
 	go.opentelemetry.io/otel/metric v1.39.0 // indirect
 	go.opentelemetry.io/otel/trace v1.39.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.7.1 // indirect
+	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/crypto v0.43.0 // indirect
+	golang.org/x/crypto v0.44.0 // indirect
 	golang.org/x/exp v0.0.0-20250808145144-a408d31f581a // indirect
 	golang.org/x/mod v0.29.0 // indirect
-	golang.org/x/net v0.46.1-0.20251013234738-63d1a5100f82 // indirect
+	golang.org/x/net v0.47.0 // indirect
 	golang.org/x/oauth2 v0.32.0 // indirect
 	golang.org/x/sys v0.39.0 // indirect
-	golang.org/x/term v0.36.0 // indirect
-	golang.org/x/text v0.30.0 // indirect
+	golang.org/x/term v0.37.0 // indirect
+	golang.org/x/text v0.31.0 // indirect
 	golang.org/x/time v0.13.0 // indirect
 	golang.org/x/tools v0.38.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20251022142026-3a174f9686a8 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
diff --git a/go.sum b/go.sum
index 84f7a49..a339544 100644
--- a/go.sum
+++ b/go.sum
@@ -149,8 +149,8 @@ github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5T
 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA=
 github.com/grafana/regexp v0.0.0-20250905093917-f7b3be9d1853 h1:cLN4IBkmkYZNnk7EAJ0BHIethd+J6LqxFNw5mSiI2bM=
 github.com/grafana/regexp v0.0.0-20250905093917-f7b3be9d1853/go.mod h1:+JKpmjMGhpgPL+rXZ5nsZieVzvarn86asRlBg4uNGnk=
-github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU=
-github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 h1:NmZ1PKzSTQbuGHw9DGPFomqkkLWMC+vZCkfs+FHv1Vg=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3/go.mod h1:zQrxl1YP88HQlA6i9c63DSVPFklWpGX4OWAc9bFuaH4=
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/huandu/xstrings v1.3.3 h1:/Gcsuc1x8JVbJ9/rlye4xZnVAbEkGauT8lbebqcQws4=
@@ -274,10 +274,10 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 h1:RbKq8BG
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0/go.mod h1:h06DGIukJOevXaj/xrNjhi/2098RZzcLTbc0jDAUbsg=
 go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
 go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 h1:GqRJVj7UmLjCVyVJ3ZFLdPRmhDUp2zFmQe3RHIOsw24=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0/go.mod h1:ri3aaHSmCTVYu2AWv44YMauwAQc0aqI9gHKIcSbI1pU=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0 h1:lwI4Dc5leUqENgGuQImwLo4WnuXFPetmPpkLi2IrX54=
-go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0/go.mod h1:Kz/oCE7z5wuyhPxsXDuaPteSWqjSBD5YaSdbxZYGbGk=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 h1:f0cb2XPmrqn4XMy9PNliTgRKJgS5WcL/u0/WRYGz4t0=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0/go.mod h1:vnakAaFckOMiMtOIhFI2MNH4FYrZzXCYxmb1LlhoGz8=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.39.0 h1:in9O8ESIOlwJAEGTkkf34DesGRAc/Pn8qJ7k3r/42LM=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.39.0/go.mod h1:Rp0EXBm5tfnv0WL+ARyO/PHBEaEAT8UUHQ6AGJcSq6c=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.39.0 h1:8UPA4IbVZxpsD76ihGOQiFml99GPAEZLohDXvqHdi6U=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.39.0/go.mod h1:MZ1T/+51uIVKlRzGw1Fo46KEWThjlCBZKl2LzY5nv4g=
 go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
@@ -288,8 +288,8 @@ go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2W
 go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
 go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
 go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
-go.opentelemetry.io/proto/otlp v1.7.1 h1:gTOMpGDb0WTBOP8JaO72iL3auEZhVmAQg4ipjOVAtj4=
-go.opentelemetry.io/proto/otlp v1.7.1/go.mod h1:b2rVh6rfI/s2pHWNlB7ILJcRALpcNDzKhACevjI+ZnE=
+go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A=
+go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
@@ -305,8 +305,8 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 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.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04=
-golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0=
+golang.org/x/crypto v0.44.0 h1:A97SsFvM3AIwEEmTBiaxPPTYpDC47w720rdiiUvgoAU=
+golang.org/x/crypto v0.44.0/go.mod h1:013i+Nw79BMiQiMsOPcVCB5ZIJbYkerPrGnOa00tvmc=
 golang.org/x/exp v0.0.0-20250808145144-a408d31f581a h1:Y+7uR/b1Mw2iSXZ3G//1haIiSElDQZ8KWh0h+sZPG90=
 golang.org/x/exp v0.0.0-20250808145144-a408d31f581a/go.mod h1:rT6SFzZ7oxADUDx58pcaKFTcZ+inxAa9fTrYx/uVYwg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -317,8 +317,8 @@ 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.46.1-0.20251013234738-63d1a5100f82 h1:6/3JGEh1C88g7m+qzzTbl3A0FtsLguXieqofVLU/JAo=
-golang.org/x/net v0.46.1-0.20251013234738-63d1a5100f82/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210=
+golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
+golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
 golang.org/x/oauth2 v0.32.0 h1:jsCblLleRMDrxMN29H3z/k1KliIvpLgCkE6R8FXXNgY=
 golang.org/x/oauth2 v0.32.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -331,12 +331,12 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
 golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
-golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q=
-golang.org/x/term v0.36.0/go.mod h1:Qu394IJq6V6dCBRgwqshf3mPF85AqzYEzofzRdZkWss=
+golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU=
+golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254=
 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.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k=
-golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM=
+golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM=
+golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM=
 golang.org/x/time v0.13.0 h1:eUlYslOIt32DgYD6utsuUeHs4d7AsEYLuIAdg7FlYgI=
 golang.org/x/time v0.13.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -359,10 +359,10 @@ gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
 gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
 google.golang.org/api v0.250.0 h1:qvkwrf/raASj82UegU2RSDGWi/89WkLckn4LuO4lVXM=
 google.golang.org/api v0.250.0/go.mod h1:Y9Uup8bDLJJtMzJyQnu+rLRJLA0wn+wTtc6vTlOvfXo=
-google.golang.org/genproto/googleapis/api v0.0.0-20251022142026-3a174f9686a8 h1:mepRgnBZa07I4TRuomDE4sTIYieg/osKmzIf4USdWS4=
-google.golang.org/genproto/googleapis/api v0.0.0-20251022142026-3a174f9686a8/go.mod h1:fDMmzKV90WSg1NbozdqrE64fkuTv6mlq2zxo9ad+3yo=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8 h1:M1rk8KBnUsBDg1oPGHNCxG4vc1f49epmTO7xscSajMk=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 h1:fCvbg86sFXwdrl5LgVcTEvNC+2txB5mgROGmRL5mrls=
+google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
 google.golang.org/grpc v1.77.0 h1:wVVY6/8cGA6vvffn+wWK5ToddbgdU3d8MNENr4evgXM=
 google.golang.org/grpc v1.77.0/go.mod h1:z0BY1iVj0q8E1uSQCjL9cppRj+gnZjzDnzV0dHhrNig=
 google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
diff --git a/pkg/epp/flowcontrol/config.go b/pkg/epp/flowcontrol/config.go
index 9c3cc87..761b29b 100644
--- a/pkg/epp/flowcontrol/config.go
+++ b/pkg/epp/flowcontrol/config.go
@@ -32,7 +32,7 @@ const (
 // and initialization.
 type Config struct {
 	Controller controller.Config
-	Registry   registry.Config
+	Registry   *registry.Config
 }
 
 // ValidateAndApplyDefaults checks the configuration for validity and populates any empty fields with system defaults.
@@ -43,12 +43,8 @@ func (c *Config) ValidateAndApplyDefaults() (*Config, error) {
 	if err != nil {
 		return nil, fmt.Errorf("controller config validation failed: %w", err)
 	}
-	validatedRegistryCfg, err := c.Registry.ValidateAndApplyDefaults()
-	if err != nil {
-		return nil, fmt.Errorf("registry config validation failed: %w", err)
-	}
 	return &Config{
 		Controller: *validatedControllerCfg,
-		Registry:   *validatedRegistryCfg,
+		Registry:   c.Registry,
 	}, nil
 }
diff --git a/pkg/epp/flowcontrol/registry/config.go b/pkg/epp/flowcontrol/registry/config.go
index 6ded14d..d74a4da 100644
--- a/pkg/epp/flowcontrol/registry/config.go
+++ b/pkg/epp/flowcontrol/registry/config.go
@@ -29,9 +29,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
 )
 
-// =============================================================================
-// Default Values
-// =============================================================================
+// --- Defaults ---
 
 const (
 	// defaultPriorityBandMaxBytes is the default global capacity for a priority band if not explicitly configured.
@@ -52,15 +50,59 @@ const (
 	defaultEventChannelBufferSize int = 4096
 )
 
-// =============================================================================
-// Global Configuration (FlowRegistry)
-// =============================================================================
+// --- Capability Checking ---
 
-// Config holds the master configuration for the entire `FlowRegistry`.
+// capabilityChecker abstracts the logic required to validate if a policy is compatible with a queue.
+type capabilityChecker interface {
+	CheckCompatibility(p intra.RegisteredPolicyName, q queue.RegisteredQueueName) error
+}
+
+// runtimeCapabilityChecker is the default implementation used in production.
+// It instantiates the actual plugins to inspect their required and provided capabilities.
+type runtimeCapabilityChecker struct{}
+
+func (r *runtimeCapabilityChecker) CheckCompatibility(p intra.RegisteredPolicyName, q queue.RegisteredQueueName) error {
+	tempPolicy, err := intra.NewPolicyFromName(p)
+	if err != nil {
+		return fmt.Errorf("failed to validate policy %q: %w", p, err)
+	}
+
+	requiredCapabilities := tempPolicy.RequiredQueueCapabilities()
+
+	// We pass nil for the comparator as we only need to inspect static capabilities here.
+	tempQueue, err := queue.NewQueueFromName(q, nil)
+	if err != nil {
+		return fmt.Errorf("failed to instantiate queue type %q: %w", q, err)
+	}
+
+	if len(requiredCapabilities) == 0 {
+		return nil // The policy is compatible with any queue type.
+	}
+
+	queueCapabilities := tempQueue.Capabilities()
+	capabilitySet := make(map[framework.QueueCapability]struct{}, len(queueCapabilities))
+	for _, cap := range queueCapabilities {
+		capabilitySet[cap] = struct{}{}
+	}
+
+	for _, req := range requiredCapabilities {
+		if _, ok := capabilitySet[req]; !ok {
+			return fmt.Errorf(
+				"policy %q is not compatible with queue %q: missing capability %q: %w",
+				tempPolicy.Name(),
+				tempQueue.Name(),
+				req,
+				contracts.ErrPolicyQueueIncompatible,
+			)
+		}
+	}
+	return nil
+}
+
+// --- Configuration ---
+
+// Config holds the master configuration for the entire FlowRegistry.
 // It serves as the top-level blueprint, defining global settings and the templates for its priority bands.
-//
-// This configuration is validated and defaulted once at startup. It is then used to generate partitioned `ShardConfig`
-// instances for each internal `registryShard`.
 type Config struct {
 	// MaxBytes defines an optional, global maximum total byte size limit aggregated across all priority bands and shards.
 	// The `controller.FlowController` enforces this limit in addition to per-band capacity limits.
@@ -69,10 +111,9 @@ type Config struct {
 	MaxBytes uint64
 
 	// PriorityBands defines the set of priority band templates managed by the `FlowRegistry`.
-	// The configuration for each band, including its default policies and queue types, is specified here.
-	// These templates are used to generate partitioned `ShardPriorityBandConfig`s.
+	// It is a map keyed by Priority level, providing O(1) access and ensuring priority uniqueness by definition.
 	// Required: At least one `PriorityBandConfig` must be provided for a functional registry.
-	PriorityBands []PriorityBandConfig
+	PriorityBands map[int]*PriorityBandConfig
 
 	// InitialShardCount specifies the number of parallel shards to create when the registry is initialized.
 	// This value must be greater than zero.
@@ -90,19 +131,6 @@ type Config struct {
 	// This value must be greater than zero.
 	// Optional: Defaults to `defaultEventChannelBufferSize` (4096).
 	EventChannelBufferSize int
-
-	// priorityBandMap provides O(1) lookups of `PriorityBandConfig` by priority level.
-	// Crucially, this map serves as a correctness mechanism. It ensures that accessors return a safe, stable pointer to
-	// the correct element within this specific configuration instance, preventing common "pointer-to-loop-variable"
-	// errors, especially across deep copies or partitioning.
-	// It is populated during validation and when the config is copied or partitioned.
-	priorityBandMap map[int]*PriorityBandConfig
-
-	// Factory functions used for plugin instantiation during configuration validation.
-	// These enable dependency injection for unit testing the validation logic.
-	interFlowDispatchPolicyFactory interFlowDispatchPolicyFactory
-	intraFlowDispatchPolicyFactory intraFlowDispatchPolicyFactory
-	queueFactory                   queueFactory
 }
 
 // PriorityBandConfig defines the configuration template for a single priority band.
@@ -121,203 +149,281 @@ type PriorityBandConfig struct {
 
 	// IntraFlowDispatchPolicy specifies the default name of the policy used to select a request from within a single
 	// flow's queue in this band.
-	// Optional: Defaults to `defaultIntraFlowDispatchPolicy` ("FCFS").
+	// Optional: Defaults to defaultIntraFlowDispatchPolicy ("FCFS").
 	IntraFlowDispatchPolicy intra.RegisteredPolicyName
 
 	// InterFlowDispatchPolicy specifies the name of the policy used to select which flow's queue to service next from
 	// this band.
-	// Optional: Defaults to `defaultInterFlowDispatchPolicy` ("BestHead").
+	// Optional: Defaults to defaultInterFlowDispatchPolicy ("BestHead").
 	InterFlowDispatchPolicy interflow.RegisteredPolicyName
 
 	// Queue specifies the default name of the `framework.SafeQueue` implementation for flow queues in this band.
-	// Optional: Defaults to `defaultQueue` ("ListQueue").
+	// Optional: Defaults to defaultQueue ("ListQueue").
 	Queue queue.RegisteredQueueName
 
 	// MaxBytes defines the maximum total byte size for this priority band, aggregated across all shards.
-	// Optional: Defaults to `defaultPriorityBandMaxBytes` (1 GB).
+	// Optional: Defaults to defaultPriorityBandMaxBytes (1 GB).
 	MaxBytes uint64
 }
 
-// =============================================================================
-// Shard-Level Configuration
-// =============================================================================
+// --- Config Functional Options ---
 
-// ShardConfig holds the partitioned configuration for a single `registryShard`.
-// It contains only the data relevant to that shard and is derived from the global `Config`.
-type ShardConfig struct {
-	// MaxBytes is this shard's partitioned portion of the global `Config.MaxBytes` limit.
-	MaxBytes uint64
+// configBuilder holds the intermediate state during NewConfig.
+// It allows us to manage build-time dependencies (like capabilityChecker) without polluting the final Config struct.
+type configBuilder struct {
+	config  *Config
+	checker capabilityChecker
+}
+
+// ConfigOption defines a functional option for configuring the registry.
+type ConfigOption func(*configBuilder) error
 
-	// PriorityBands holds the partitioned configuration for each priority band managed by this shard.
-	PriorityBands []ShardPriorityBandConfig
+// WithMaxBytes sets the global maximum total byte size limit.
+func WithMaxBytes(maxBytes uint64) ConfigOption {
+	return func(b *configBuilder) error {
+		b.config.MaxBytes = maxBytes
+		return nil
+	}
+}
 
-	// priorityBandMap provides O(1) lookups of `ShardPriorityBandConfig` by priority level.
-	// It serves as a correctness mechanism, ensuring that accessors return a safe, stable pointer to the correct element
-	// within this specific shard configuration instance.
-	priorityBandMap map[int]*ShardPriorityBandConfig
+// WithInitialShardCount sets the number of shards to create on startup.
+func WithInitialShardCount(count int) ConfigOption {
+	return func(b *configBuilder) error {
+		if count <= 0 {
+			return errors.New("initialShardCount must be greater than 0")
+		}
+		b.config.InitialShardCount = count
+		return nil
+	}
 }
 
-// ShardPriorityBandConfig holds the partitioned configuration for a single priority band within a single shard.
-type ShardPriorityBandConfig struct {
-	// Priority is the unique numerical priority level for this band.
-	Priority int
-	// PriorityName is a unique human-readable name for this priority band.
-	PriorityName string
-	// IntraFlowDispatchPolicy is the name of the policy for dispatch within a flow's queue.
-	IntraFlowDispatchPolicy intra.RegisteredPolicyName
-	// InterFlowDispatchPolicy is the name of the policy for dispatch between flow queues.
-	InterFlowDispatchPolicy interflow.RegisteredPolicyName
-	// Queue is the name of the queue implementation to use.
-	Queue queue.RegisteredQueueName
-	// MaxBytes is this shard's partitioned portion of this band's global capacity limit.
-	// The `controller.FlowController` enforces this limit for this specific shard.
-	MaxBytes uint64
+// WithFlowGCTimeout sets the idle flow garbage collection interval.
+func WithFlowGCTimeout(d time.Duration) ConfigOption {
+	return func(b *configBuilder) error {
+		if d <= 0 {
+			return errors.New("flowGCTimeout must be positive")
+		}
+		b.config.FlowGCTimeout = d
+		return nil
+	}
 }
 
-// getBandConfig finds and returns the shard-level configuration for a specific priority level.
-// Returns an error wrapping `contracts.ErrPriorityBandNotFound` if the priority is not configured.
-func (sc *ShardConfig) getBandConfig(priority int) (*ShardPriorityBandConfig, error) {
-	if band, ok := sc.priorityBandMap[priority]; ok {
-		return band, nil
+// WithPriorityBand adds a priority band configuration.
+// If a band with the same Priority already exists, it returns an error.
+func WithPriorityBand(band *PriorityBandConfig) ConfigOption {
+	return func(b *configBuilder) error {
+		if band == nil {
+			return errors.New("cannot add nil PriorityBandConfig")
+		}
+		if _, exists := b.config.PriorityBands[band.Priority]; exists {
+			return fmt.Errorf("duplicate priority level %d", band.Priority)
+		}
+		b.config.PriorityBands[band.Priority] = band
+		return nil
 	}
-	return nil, fmt.Errorf("shard config for priority %d not found: %w", priority, contracts.ErrPriorityBandNotFound)
 }
 
-// =============================================================================
-// Internal Methods and Helpers
-// =============================================================================
+// withCapabilityChecker overrides the compatibility checker used during validation.
+// It is intended for use only in internal unit tests.
+// test-only
+func withCapabilityChecker(checker capabilityChecker) ConfigOption {
+	return func(b *configBuilder) error {
+		if checker == nil {
+			return errors.New("cannot set nil CapabilityChecker")
+		}
+		b.checker = checker
+		return nil
+	}
+}
 
-// --- Validation and Defaulting ---
+// --- PriorityBandConfig Functional Options ---
 
-// ValidateAndApplyDefaults checks the global configuration for validity and then creates a new `Config` object,
-// populating any empty fields with system defaults.
-// It does not mutate the receiver.
-func (c *Config) ValidateAndApplyDefaults() (*Config, error) {
-	cfg := c.deepCopy()
+// PriorityBandConfigOption defines a functional option for configuring a single PriorityBandConfig.
+type PriorityBandConfigOption func(*PriorityBandConfig) error
 
-	// Apply defaults to top-level fields.
-	if cfg.InitialShardCount <= 0 {
-		cfg.InitialShardCount = defaultInitialShardCount
-	}
-	if cfg.FlowGCTimeout <= 0 {
-		cfg.FlowGCTimeout = defaultFlowGCTimeout
+// WithIntraFlowPolicy sets the intra-flow dispatch policy (e.g., "FCFS").
+func WithIntraFlowPolicy(name intra.RegisteredPolicyName) PriorityBandConfigOption {
+	return func(p *PriorityBandConfig) error {
+		if name == "" {
+			return errors.New("IntraFlowDispatchPolicy cannot be empty")
+		}
+		p.IntraFlowDispatchPolicy = name
+		return nil
 	}
-	if cfg.EventChannelBufferSize <= 0 {
-		cfg.EventChannelBufferSize = defaultEventChannelBufferSize
+}
+
+// WithInterFlowPolicy sets the inter-flow dispatch policy (e.g., "RoundRobin").
+func WithInterFlowPolicy(name interflow.RegisteredPolicyName) PriorityBandConfigOption {
+	return func(p *PriorityBandConfig) error {
+		if name == "" {
+			return errors.New("InterFlowDispatchPolicy cannot be empty")
+		}
+		p.InterFlowDispatchPolicy = name
+		return nil
 	}
+}
 
-	// Ensure the DI factories are initialized for production use if `NewConfig` was called without options.
-	if cfg.interFlowDispatchPolicyFactory == nil {
-		cfg.interFlowDispatchPolicyFactory = interflow.NewPolicyFromName
+// WithQueue sets the queue implementation (e.g., "ListQueue") for flows in this band.
+func WithQueue(name queue.RegisteredQueueName) PriorityBandConfigOption {
+	return func(p *PriorityBandConfig) error {
+		if name == "" {
+			return errors.New("Queue cannot be empty")
+		}
+		p.Queue = name
+		return nil
 	}
-	if cfg.intraFlowDispatchPolicyFactory == nil {
-		cfg.intraFlowDispatchPolicyFactory = intra.NewPolicyFromName
+}
+
+// WithBandMaxBytes sets the capacity limit for this specific priority band.
+func WithBandMaxBytes(maxBytes uint64) PriorityBandConfigOption {
+	return func(p *PriorityBandConfig) error {
+		p.MaxBytes = maxBytes
+		return nil
 	}
-	if cfg.queueFactory == nil {
-		cfg.queueFactory = queue.NewQueueFromName
+}
+
+// --- Constructors ---
+
+// NewConfig creates a new Config populated with system defaults, applies the provided options, and enforces strict
+// validation.
+func NewConfig(opts ...ConfigOption) (*Config, error) {
+	builder := &configBuilder{
+		config: &Config{
+			MaxBytes:               0, // no limit enforced
+			InitialShardCount:      defaultInitialShardCount,
+			FlowGCTimeout:          defaultFlowGCTimeout,
+			EventChannelBufferSize: defaultEventChannelBufferSize,
+			PriorityBands:          make(map[int]*PriorityBandConfig),
+		},
+		checker: &runtimeCapabilityChecker{},
 	}
 
-	if len(cfg.PriorityBands) == 0 {
-		return nil, errors.New("config validation failed: at least one priority band must be defined")
+	for _, opt := range opts {
+		if err := opt(builder); err != nil {
+			return nil, err
+		}
 	}
 
-	// Validate and default each priority band.
-	priorities := make(map[int]struct{})
-	priorityNames := make(map[string]struct{})
-	cfg.priorityBandMap = make(map[int]*PriorityBandConfig, len(cfg.PriorityBands))
+	// Apply defaults to all bands.
+	// This covers the case where a user passed a raw struct literal via WithPriorityBand.
+	for _, band := range builder.config.PriorityBands {
+		band.applyDefaults()
+	}
 
-	for i := range cfg.PriorityBands {
-		band := &cfg.PriorityBands[i]
-		if _, exists := priorities[band.Priority]; exists {
-			return nil, fmt.Errorf("config validation failed: duplicate priority level %d found", band.Priority)
-		}
-		priorities[band.Priority] = struct{}{}
+	if err := builder.config.validate(builder.checker); err != nil {
+		return nil, fmt.Errorf("invalid registry config: %w", err)
+	}
+	return builder.config, nil
+}
 
-		if band.PriorityName == "" {
-			return nil, fmt.Errorf("config validation failed: PriorityName is required for priority band %d", band.Priority)
-		}
-		if _, exists := priorityNames[band.PriorityName]; exists {
-			return nil, fmt.Errorf("config validation failed: duplicate priority name %q found", band.PriorityName)
-		}
-		priorityNames[band.PriorityName] = struct{}{}
+// NewPriorityBandConfig creates a new band configuration with the required fields.
+// It applies system defaults first, then applies any provided options to override those defaults
+func NewPriorityBandConfig(priority int, name string, opts ...PriorityBandConfigOption) (*PriorityBandConfig, error) {
+	pb := &PriorityBandConfig{
+		Priority:     priority,
+		PriorityName: name,
+	}
 
-		if band.IntraFlowDispatchPolicy == "" {
-			band.IntraFlowDispatchPolicy = defaultIntraFlowDispatchPolicy
-		}
-		if band.InterFlowDispatchPolicy == "" {
-			band.InterFlowDispatchPolicy = defaultInterFlowDispatchPolicy
-		}
-		if band.Queue == "" {
-			band.Queue = defaultQueue
-		}
-		if band.MaxBytes == 0 {
-			band.MaxBytes = defaultPriorityBandMaxBytes
-		}
+	pb.applyDefaults()
 
-		if err := cfg.validateBandCompatibility(*band); err != nil {
+	for _, opt := range opts {
+		if err := opt(pb); err != nil {
 			return nil, err
 		}
-		cfg.priorityBandMap[band.Priority] = band
 	}
-	return cfg, nil
+
+	return pb, nil
 }
 
-// validateBandCompatibility verifies that a band's configured queue type has the necessary capabilities.
-func (c *Config) validateBandCompatibility(band PriorityBandConfig) error {
-	policy, err := c.intraFlowDispatchPolicyFactory(band.IntraFlowDispatchPolicy)
-	if err != nil {
-		return fmt.Errorf("failed to validate policy %q for priority band %d: %w",
-			band.IntraFlowDispatchPolicy, band.Priority, err)
+// --- Validation & Defaults ---
+
+func (p *PriorityBandConfig) applyDefaults() {
+	if p.IntraFlowDispatchPolicy == "" {
+		p.IntraFlowDispatchPolicy = defaultIntraFlowDispatchPolicy
 	}
+	if p.InterFlowDispatchPolicy == "" {
+		p.InterFlowDispatchPolicy = defaultInterFlowDispatchPolicy
+	}
+	if p.Queue == "" {
+		p.Queue = defaultQueue
+	}
+	if p.MaxBytes == 0 {
+		p.MaxBytes = defaultPriorityBandMaxBytes
+	}
+}
 
-	requiredCapabilities := policy.RequiredQueueCapabilities()
-	if len(requiredCapabilities) == 0 {
-		return nil // The policy is compatible with any queue type.
+// validate checks the integrity of a single band's configuration.
+func (p *PriorityBandConfig) validate(checker capabilityChecker) error {
+	if p.PriorityName == "" {
+		return fmt.Errorf("PriorityName is required for priority band %d", p.Priority)
+	}
+	if p.IntraFlowDispatchPolicy == "" {
+		return fmt.Errorf("IntraFlowDispatchPolicy required for priority band %d", p.Priority)
+	}
+	if p.InterFlowDispatchPolicy == "" {
+		return fmt.Errorf("InterFlowDispatchPolicy required for priority band %d", p.Priority)
 	}
+	if p.Queue == "" {
+		return fmt.Errorf("Queue required for priority band %d", p.Priority)
+	}
+	if checker != nil {
+		if err := checker.CheckCompatibility(p.IntraFlowDispatchPolicy, p.Queue); err != nil {
+			return fmt.Errorf("priority band %d (%s) configuration error: %w",
+				p.Priority, p.PriorityName, err)
+		}
+	}
+	return nil
+}
 
-	// Create a temporary queue instance to inspect its capabilities.
-	tempQueue, err := c.queueFactory(band.Queue, nil)
-	if err != nil {
-		return fmt.Errorf("failed to instantiate queue type %q for capability inspection (priority band %d): %w",
-			band.Queue, band.Priority, err)
+// validate checks global constraints and delegates band validation.
+func (c *Config) validate(checker capabilityChecker) error {
+	if c.InitialShardCount <= 0 {
+		return errors.New("initialShardCount must be greater than 0")
+	}
+	if c.FlowGCTimeout <= 0 {
+		return errors.New("flowGCTimeout must be positive")
+	}
+	if c.EventChannelBufferSize <= 0 {
+		return errors.New("eventChannelBufferSize must be greater than 0")
 	}
-	queueCapabilities := tempQueue.Capabilities()
 
-	capabilitySet := make(map[framework.QueueCapability]struct{}, len(queueCapabilities))
-	for _, cap := range queueCapabilities {
-		capabilitySet[cap] = struct{}{}
+	if len(c.PriorityBands) == 0 {
+		return errors.New("at least one priority band must be defined")
 	}
 
-	for _, req := range requiredCapabilities {
-		if _, ok := capabilitySet[req]; !ok {
-			return fmt.Errorf(
-				"policy %q is not compatible with queue %q for priority band %d (%s): missing capability %q: %w",
-				policy.Name(),
-				tempQueue.Name(),
-				band.Priority,
-				band.PriorityName,
-				req,
-				contracts.ErrPolicyQueueIncompatible,
-			)
+	names := make(map[string]struct{}, len(c.PriorityBands))
+	for _, band := range c.PriorityBands {
+		if _, exists := names[band.PriorityName]; exists {
+			return fmt.Errorf("duplicate priority name %q found", band.PriorityName)
+		}
+		names[band.PriorityName] = struct{}{}
+
+		if err := band.validate(checker); err != nil {
+			return err
 		}
 	}
 	return nil
 }
 
-// --- Partitioning Logic ---
+// --- Sharding & Partitioning ---
+
+// ShardConfig holds the partitioned configuration for a single registryShard.
+type ShardConfig struct {
+	MaxBytes      uint64
+	PriorityBands map[int]*PriorityBandConfig
+}
 
 // partition derives a `ShardConfig` from the master `Config` for a specific shard index.
 // It calculates the capacity distribution, ensuring that the total global capacity is distributed completely and
 // evenly.
 func (c *Config) partition(shardIndex, totalShards int) *ShardConfig {
 	shardCfg := &ShardConfig{
-		MaxBytes:        partitionUint64(c.MaxBytes, shardIndex, totalShards),
-		PriorityBands:   make([]ShardPriorityBandConfig, len(c.PriorityBands)),
-		priorityBandMap: make(map[int]*ShardPriorityBandConfig, len(c.PriorityBands)),
+		MaxBytes:      partitionUint64(c.MaxBytes, shardIndex, totalShards),
+		PriorityBands: make(map[int]*PriorityBandConfig, len(c.PriorityBands)),
 	}
 
-	for i, template := range c.PriorityBands {
-		shardBandCfg := ShardPriorityBandConfig{
+	for _, template := range c.PriorityBands {
+		shardBand := &PriorityBandConfig{
 			Priority:                template.Priority,
 			PriorityName:            template.PriorityName,
 			IntraFlowDispatchPolicy: template.IntraFlowDispatchPolicy,
@@ -325,11 +431,8 @@ func (c *Config) partition(shardIndex, totalShards int) *ShardConfig {
 			Queue:                   template.Queue,
 			MaxBytes:                partitionUint64(template.MaxBytes, shardIndex, totalShards),
 		}
-		shardCfg.PriorityBands[i] = shardBandCfg
 
-		// Crucial: We must take the address of the element within the new slice (`shardCfg.PriorityBands`) to ensure the
-		// map pointers are correct for the newly created ShardConfig instance.
-		shardCfg.priorityBandMap[shardBandCfg.Priority] = &shardCfg.PriorityBands[i]
+		shardCfg.PriorityBands[shardBand.Priority] = shardBand
 	}
 	return shardCfg
 }
@@ -337,129 +440,36 @@ func (c *Config) partition(shardIndex, totalShards int) *ShardConfig {
 // partitionUint64 distributes a total uint64 value across a number of partitions.
 // It distributes the remainder of the division one by one to the first few partitions.
 func partitionUint64(total uint64, partitionIndex, totalPartitions int) uint64 {
-	if totalPartitions <= 0 {
-		panic(fmt.Sprintf("invariant violation: cannot partition into %d partitions", totalPartitions))
-	}
 	if total == 0 {
 		return 0
 	}
-	base := total / uint64(totalPartitions)
-	remainder := total % uint64(totalPartitions)
-	// Distribute the remainder. The first `remainder` partitions get one extra from the total.
-	// For example, if total=10 and partitions=3, base=3, remainder=1. Partition 0 gets 3+1=4, partitions 1 and 2 get 3.
-	if uint64(partitionIndex) < remainder {
-		base++
-	}
-	return base
-}
-
-// --- Test-Only Dependency Injection ---
-
-// configOption is a function that applies a configuration change to a `Config` object.
-// This is used exclusively for dependency injection in tests and should not be used in production code.
-type configOption func(*Config)
-
-// interFlowDispatchPolicyFactory defines the signature for a function that creates an
-// `framework.InterFlowDispatchPolicy` instance from its registered name.
-// It serves as an abstraction over the concrete `interflow.NewPolicyFromName` factory, enabling dependency injection for
-// testing validation logic.
-type interFlowDispatchPolicyFactory func(name interflow.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error)
-
-// intraFlowDispatchPolicyFactory defines the signature for a function that creates an
-// `framework.IntraFlowDispatchPolicy` instance from its registered name.
-// It serves as an abstraction over the concrete `intra.NewPolicyFromName` factory, enabling dependency injection for
-// testing validation logic.
-type intraFlowDispatchPolicyFactory func(name intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error)
-
-// queueFactory defines the signature for a function that creates a `framework.SafeQueue` instance from its registered
-// name and a given `framework.ItemComparator`.
-// It serves as an abstraction over the concrete `queue.NewQueueFromName` factory, enabling dependency injection for
-// testing validation logic.
-type queueFactory func(
-	name queue.RegisteredQueueName,
-	comparator framework.ItemComparator,
-) (framework.SafeQueue, error)
-
-// withInterFlowDispatchPolicyFactory returns a test-only `configOption` that overrides the factory function for
-// creating `framework.InterFlowDispatchPolicy` instances.
-// This is used exclusively for testing validation logic.
-// test-only
-func withInterFlowDispatchPolicyFactory(factory interFlowDispatchPolicyFactory) configOption {
-	return func(c *Config) {
-		c.interFlowDispatchPolicyFactory = factory
-	}
-}
-
-// withIntraFlowDispatchPolicyFactory returns a test-only `configOption` that overrides the factory function for
-// creating `framework.IntraFlowDispatchPolicy` instances.
-// This is used exclusively for testing validation logic.
-// test-only
-func withIntraFlowDispatchPolicyFactory(factory intraFlowDispatchPolicyFactory) configOption {
-	return func(c *Config) {
-		c.intraFlowDispatchPolicyFactory = factory
-	}
-}
 
-// withQueueFactory returns a test-only `configOption` that overrides the factory function for creating
-// `framework.SafeQueue` instances.
-// This is used exclusively for testing validation logic.
-// test-only
-func withQueueFactory(factory queueFactory) configOption {
-	return func(c *Config) {
-		c.queueFactory = factory
-	}
-}
+	t := uint64(totalPartitions)
+	base := total / t
+	remainder := total % t
 
-// newConfig creates a new validated and defaulted `Config` object.
-// It applies provided test-only functional options before validation and defaulting.
-// It does not mutate the input `cfg`.
-// test-only
-func newConfig(cfg Config, opts ...configOption) (*Config, error) {
-	newCfg := cfg.deepCopy()
-	for _, opt := range opts {
-		opt(newCfg)
+	if uint64(partitionIndex) < remainder {
+		return base + 1
 	}
-	return newCfg.ValidateAndApplyDefaults()
+	return base
 }
 
-// --- Internal Utilities ---
-
-// deepCopy creates a deep copy of the `Config` object.
-func (c *Config) deepCopy() *Config {
+// Clone creates a deep copy of the Config.
+// It ensures the new Config has its own independent map and PriorityBandConfig instances.
+func (c *Config) Clone() *Config {
 	if c == nil {
 		return nil
 	}
-	newCfg := &Config{
-		MaxBytes:                       c.MaxBytes,
-		InitialShardCount:              c.InitialShardCount,
-		FlowGCTimeout:                  c.FlowGCTimeout,
-		EventChannelBufferSize:         c.EventChannelBufferSize,
-		PriorityBands:                  make([]PriorityBandConfig, len(c.PriorityBands)),
-		interFlowDispatchPolicyFactory: c.interFlowDispatchPolicyFactory,
-		intraFlowDispatchPolicyFactory: c.intraFlowDispatchPolicyFactory,
-		queueFactory:                   c.queueFactory,
-	}
-
-	// PriorityBandConfig contains only value types, so a slice copy is sufficient for a deep copy.
-	copy(newCfg.PriorityBands, c.PriorityBands)
-
-	if c.priorityBandMap != nil {
-		newCfg.priorityBandMap = make(map[int]*PriorityBandConfig, len(c.PriorityBands))
-		// Crucial: We must rebuild the map and take the address of the elements within the new slice (`newCfg.PriorityBands`)
-		// to ensure the map pointers are correct for the newly created `Config` instance.
-		for i := range newCfg.PriorityBands {
-			band := &newCfg.PriorityBands[i]
-			newCfg.priorityBandMap[band.Priority] = band
-		}
-	}
-	return newCfg
-}
 
-// getBandConfig finds and returns the global configuration template for a specific priority level.
-// Returns an error wrapping `contracts.ErrPriorityBandNotFound` if the priority is not configured.
-func (c *Config) getBandConfig(priority int) (*PriorityBandConfig, error) {
-	if band, ok := c.priorityBandMap[priority]; ok {
-		return band, nil
+	clone := *c
+	if c.PriorityBands != nil {
+		clone.PriorityBands = make(map[int]*PriorityBandConfig, len(c.PriorityBands))
+		for prio, band := range c.PriorityBands {
+			// Dereference the pointer to copy the struct value, then take the address of the new value.
+			// This ensures 'clone' points to a new memory address.
+			b := *band
+			clone.PriorityBands[prio] = &b
+		}
 	}
-	return nil, fmt.Errorf("global config for priority %d not found: %w", priority, contracts.ErrPriorityBandNotFound)
+	return &clone
 }
diff --git a/pkg/epp/flowcontrol/registry/registry.go b/pkg/epp/flowcontrol/registry/registry.go
index 3a73ef7..087cd37 100644
--- a/pkg/epp/flowcontrol/registry/registry.go
+++ b/pkg/epp/flowcontrol/registry/registry.go
@@ -30,6 +30,8 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+	intra "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
@@ -147,8 +149,8 @@ func withClock(clk clock.WithTickerAndDelayedExecution) RegistryOption {
 }
 
 // NewFlowRegistry creates and initializes a new `FlowRegistry` instance.
-func NewFlowRegistry(config Config, logger logr.Logger, opts ...RegistryOption) (*FlowRegistry, error) {
-	cfg := config.deepCopy()
+func NewFlowRegistry(config *Config, logger logr.Logger, opts ...RegistryOption) (*FlowRegistry, error) {
+	cfg := config.Clone()
 	fr := &FlowRegistry{
 		config:               cfg,
 		logger:               logger.WithName("flow-registry"),
@@ -164,9 +166,8 @@ func NewFlowRegistry(config Config, logger logr.Logger, opts ...RegistryOption)
 		fr.clock = &clock.RealClock{}
 	}
 
-	for i := range config.PriorityBands {
-		band := &config.PriorityBands[i]
-		fr.perPriorityBandStats[band.Priority] = &bandStats{}
+	for prio := range cfg.PriorityBands {
+		fr.perPriorityBandStats[prio] = &bandStats{}
 	}
 
 	if err := fr.updateShardCount(cfg.InitialShardCount); err != nil {
@@ -277,7 +278,6 @@ func (fr *FlowRegistry) prepareNewFlow(key types.FlowKey) (*flowState, error) {
 //
 // Statistics are aggregated using high-performance, lock-free atomic updates.
 // The returned stats represent a near-consistent snapshot of the system's state.
-// It is not perfectly atomic because the various counters are loaded independently without a global lock.
 func (fr *FlowRegistry) Stats() contracts.AggregateStats {
 	// Casts from `int64` to `uint64` are safe because the non-negativity invariant is strictly enforced at the
 	// `managedQueue` level.
@@ -289,10 +289,7 @@ func (fr *FlowRegistry) Stats() contracts.AggregateStats {
 	}
 
 	for p, s := range fr.perPriorityBandStats {
-		bandCfg, err := fr.config.getBandConfig(p)
-		if err != nil {
-			panic(fmt.Sprintf("invariant violation: priority band config (%d) missing during stats aggregation: %v", p, err))
-		}
+		bandCfg := fr.config.PriorityBands[p]
 		stats.PerPriorityBandStats[p] = contracts.PriorityBandStats{
 			Priority:      p,
 			PriorityName:  bandCfg.PriorityName,
@@ -465,7 +462,6 @@ func (fr *FlowRegistry) executeScaleUpLocked(newTotalActive int) error {
 			partitionedConfig,
 			fr.logger,
 			fr.propagateStatsDelta,
-			fr.config.interFlowDispatchPolicyFactory,
 		)
 		if err != nil {
 			return fmt.Errorf("failed to create new shard object %s: %w", shardID, err)
@@ -544,19 +540,19 @@ type flowComponents struct {
 // buildFlowComponents instantiates the necessary plugin components for a new flow instance.
 // It creates a distinct instance of each component for each shard to ensure state isolation.
 func (fr *FlowRegistry) buildFlowComponents(key types.FlowKey, numInstances int) ([]flowComponents, error) {
-	bandConfig, err := fr.config.getBandConfig(key.Priority)
-	if err != nil {
-		return nil, err
+	bandConfig, ok := fr.config.PriorityBands[key.Priority]
+	if !ok {
+		return nil, fmt.Errorf("priority band %d not found: %w", key.Priority, contracts.ErrPriorityBandNotFound)
 	}
 
 	allComponents := make([]flowComponents, numInstances)
 	for i := range numInstances {
-		policy, err := fr.config.intraFlowDispatchPolicyFactory(bandConfig.IntraFlowDispatchPolicy)
+		policy, err := intra.NewPolicyFromName(bandConfig.IntraFlowDispatchPolicy)
 		if err != nil {
 			return nil, fmt.Errorf("failed to instantiate intra-flow policy %q for flow %s: %w",
 				bandConfig.IntraFlowDispatchPolicy, key, err)
 		}
-		q, err := fr.config.queueFactory(bandConfig.Queue, policy.Comparator())
+		q, err := queue.NewQueueFromName(bandConfig.Queue, policy.Comparator())
 		if err != nil {
 			return nil, fmt.Errorf("failed to instantiate queue %q for flow %s: %w",
 				bandConfig.Queue, key, err)
diff --git a/pkg/epp/flowcontrol/registry/shard.go b/pkg/epp/flowcontrol/registry/shard.go
index 4a7918e..920c267 100644
--- a/pkg/epp/flowcontrol/registry/shard.go
+++ b/pkg/epp/flowcontrol/registry/shard.go
@@ -26,14 +26,19 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/interflow"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
 // priorityBand holds all `managedQueues` and configuration for a single priority level within a shard.
 type priorityBand struct {
+	// config points to the current partitioned configuration for this band.
+	// It is updated during dynamic scaling events (updateConfig), protected by the parent shard's mutex.
+	config *PriorityBandConfig
+
 	// --- Immutable (set at construction) ---
-	config                  ShardPriorityBandConfig
+
 	interFlowDispatchPolicy framework.InterFlowDispatchPolicy
 
 	// --- State Protected by the parent shard's `mu` ---
@@ -108,19 +113,19 @@ func newShard(
 	config *ShardConfig,
 	logger logr.Logger,
 	onStatsDelta propagateStatsDeltaFunc,
-	interFlowFactory interFlowDispatchPolicyFactory,
 ) (*registryShard, error) {
 	shardLogger := logger.WithName("registry-shard").WithValues("shardID", id)
 	s := &registryShard{
-		id:            id,
-		logger:        shardLogger,
-		config:        config,
-		onStatsDelta:  onStatsDelta,
-		priorityBands: make(map[int]*priorityBand, len(config.PriorityBands)),
+		id:                    id,
+		logger:                shardLogger,
+		config:                config,
+		onStatsDelta:          onStatsDelta,
+		priorityBands:         make(map[int]*priorityBand, len(config.PriorityBands)),
+		orderedPriorityLevels: make([]int, 0, len(config.PriorityBands)),
 	}
 
 	for _, bandConfig := range config.PriorityBands {
-		interPolicy, err := interFlowFactory(bandConfig.InterFlowDispatchPolicy)
+		interPolicy, err := interflow.NewPolicyFromName(bandConfig.InterFlowDispatchPolicy)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create inter-flow policy %q for priority band %d: %w",
 				bandConfig.InterFlowDispatchPolicy, bandConfig.Priority, err)
@@ -310,15 +315,8 @@ func (s *registryShard) updateConfig(newConfig *ShardConfig) {
 	defer s.mu.Unlock()
 
 	s.config = newConfig
-	// Update the partitioned config for each priority band as well.
 	for priority, band := range s.priorityBands {
-		newBandConfig, err := newConfig.getBandConfig(priority)
-		if err != nil {
-			// An invariant was violated: a priority exists in the shard but not in the new config.
-			panic(fmt.Errorf("invariant violation: priority band (%d) missing in new shard configuration during update: %w",
-				priority, err))
-		}
-		band.config = *newBandConfig
+		band.config = newConfig.PriorityBands[priority]
 	}
 	s.logger.Info("Shard configuration updated")
 }



Test patch:
diff --git a/pkg/epp/flowcontrol/config_test.go b/pkg/epp/flowcontrol/config_test.go
index 713abee..e5b959d 100644
--- a/pkg/epp/flowcontrol/config_test.go
+++ b/pkg/epp/flowcontrol/config_test.go
@@ -30,11 +30,10 @@ func TestConfig_ValidateAndApplyDefaults(t *testing.T) {
 	t.Parallel()
 
 	// A minimal valid registry config, which is required for the success case.
-	validRegistryConfig := registry.Config{
-		PriorityBands: []registry.PriorityBandConfig{
-			{Priority: 1, PriorityName: "TestBand"},
-		},
-	}
+	validRegistryConfig, err := registry.NewConfig(registry.WithPriorityBand(
+		&registry.PriorityBandConfig{Priority: 1, PriorityName: "TestBand"},
+	))
+	require.NoError(t, err)
 
 	testCases := []struct {
 		name          string
@@ -60,16 +59,6 @@ func TestConfig_ValidateAndApplyDefaults(t *testing.T) {
 			},
 			expectErr: true,
 		},
-		{
-			name: "ShouldFail_WhenRegistryConfigIsInvalid",
-			input: Config{
-				Controller: controller.Config{},
-				Registry: registry.Config{
-					PriorityBands: []registry.PriorityBandConfig{},
-				},
-			},
-			expectErr: true,
-		},
 	}
 
 	for _, tc := range testCases {
diff --git a/pkg/epp/flowcontrol/registry/config_test.go b/pkg/epp/flowcontrol/registry/config_test.go
index 05d8e57..0d47612 100644
--- a/pkg/epp/flowcontrol/registry/config_test.go
+++ b/pkg/epp/flowcontrol/registry/config_test.go
@@ -17,8 +17,6 @@ limitations under the License.
 package registry
 
 import (
-	"errors"
-	"fmt"
 	"testing"
 	"time"
 
@@ -26,485 +24,314 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/mocks"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/interflow"
 	intra "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch/fcfs"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
 )
 
-func TestConfig_ValidateAndApplyDefaults(t *testing.T) {
+// mockCapabilityChecker is a test double for verifying that NewConfig correctly delegates compatibility checks.
+type mockCapabilityChecker struct {
+	checkCompatibilityFunc func(intra intra.RegisteredPolicyName, q queue.RegisteredQueueName) error
+}
+
+func (m *mockCapabilityChecker) CheckCompatibility(p intra.RegisteredPolicyName, q queue.RegisteredQueueName) error {
+	if m.checkCompatibilityFunc != nil {
+		return m.checkCompatibilityFunc(p, q)
+	}
+	return nil
+}
+
+// mustBand is a helper to simplify test table setup.
+// It panics if the band config creation fails, which should not happen with valid static inputs.
+func mustBand(t *testing.T, priority int, name string, opts ...PriorityBandConfigOption) *PriorityBandConfig {
+	pb, err := NewPriorityBandConfig(priority, name, opts...)
+	require.NoError(t, err, "failed to create test band")
+	return pb
+}
+
+func TestNewConfig(t *testing.T) {
 	t.Parallel()
 
 	testCases := []struct {
 		name          string
-		input         Config
-		opts          []configOption
+		opts          []ConfigOption
 		expectErr     bool
-		expectedErrIs error
-		assertion     func(*testing.T, Config, *Config)
+		expectedErrIs error // Optional: check for specific wrapped error
+		assertion     func(*testing.T, *Config)
 	}{
 		// --- Success Paths ---
 		{
-			name: "ShouldApplyDefaults_WhenFieldsAreUnspecified",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{
-					{Priority: 1, PriorityName: "High"},
-					{Priority: 2, PriorityName: "Low", MaxBytes: 500},
-				},
+			name: "ShouldApplySystemDefaults_WhenNoOptionsProvided",
+			opts: []ConfigOption{
+				WithPriorityBand(mustBand(t, 1, "Default")),
 			},
-			assertion: func(t *testing.T, _ Config, newCfg *Config) {
-				assert.Equal(t, defaultInitialShardCount, newCfg.InitialShardCount, "InitialShardCount should be defaulted")
-				assert.Equal(t, defaultFlowGCTimeout, newCfg.FlowGCTimeout, "FlowGCTimeout should be defaulted")
-				assert.Equal(t, defaultEventChannelBufferSize, newCfg.EventChannelBufferSize,
+			assertion: func(t *testing.T, cfg *Config) {
+				assert.Equal(t, defaultInitialShardCount, cfg.InitialShardCount, "InitialShardCount should be defaulted")
+				assert.Equal(t, defaultFlowGCTimeout, cfg.FlowGCTimeout, "FlowGCTimeout should be defaulted")
+				assert.Equal(t, defaultEventChannelBufferSize, cfg.EventChannelBufferSize,
 					"EventChannelBufferSize should be defaulted")
-				require.Len(t, newCfg.PriorityBands, 2, "Should have two priority bands")
-				band1 := newCfg.PriorityBands[0]
-				assert.Equal(t, defaultIntraFlowDispatchPolicy, band1.IntraFlowDispatchPolicy,
-					"Band 1 IntraFlowDispatchPolicy should be defaulted")
-				assert.Equal(t, defaultInterFlowDispatchPolicy, band1.InterFlowDispatchPolicy,
-					"Band 1 InterFlowDispatchPolicy should be defaulted")
-				assert.Equal(t, defaultQueue, band1.Queue, "Band 1 Queue should be defaulted")
-				assert.Equal(t, defaultPriorityBandMaxBytes, band1.MaxBytes, "Band 1 MaxBytes should be defaulted")
-				band2 := newCfg.PriorityBands[1]
-				assert.Equal(t, uint64(500), band2.MaxBytes, "Band 2 MaxBytes should retain its specified value")
-			},
-		},
-		{
-			name: "ShouldRemainUnchanged_WhenConfigIsFullySpecifiedAndValid",
-			input: Config{
-				MaxBytes:               1000,
-				InitialShardCount:      10,
-				FlowGCTimeout:          10 * time.Minute,
-				EventChannelBufferSize: 5000,
-				PriorityBands: []PriorityBandConfig{{
-					Priority:                1,
-					PriorityName:            "Critical",
-					IntraFlowDispatchPolicy: fcfs.FCFSPolicyName,
-					InterFlowDispatchPolicy: interflow.BestHeadPolicyName,
-					Queue:                   queue.ListQueueName,
-					MaxBytes:                500,
-				}},
-			},
-			assertion: func(t *testing.T, originalCfg Config, newCfg *Config) {
-				assert.Equal(t, originalCfg.MaxBytes, newCfg.MaxBytes, "MaxBytes should remain unchanged")
-				assert.Equal(t, originalCfg.InitialShardCount, newCfg.InitialShardCount,
-					"InitialShardCount should remain unchanged")
-				assert.Equal(t, originalCfg.FlowGCTimeout, newCfg.FlowGCTimeout, "FlowGCTimeout should remain unchanged")
-				assert.Equal(t, originalCfg.EventChannelBufferSize, newCfg.EventChannelBufferSize,
-					"EventChannelBufferSize should remain unchanged")
-				assert.Equal(t, originalCfg.PriorityBands, newCfg.PriorityBands, "PriorityBands should remain unchanged")
+
+				// Verify Band Defaults
+				require.Contains(t, cfg.PriorityBands, 1)
+				band := cfg.PriorityBands[1]
+				assert.Equal(t, defaultIntraFlowDispatchPolicy, band.IntraFlowDispatchPolicy)
+				assert.Equal(t, defaultInterFlowDispatchPolicy, band.InterFlowDispatchPolicy)
+				assert.Equal(t, defaultQueue, band.Queue)
+				assert.Equal(t, defaultPriorityBandMaxBytes, band.MaxBytes)
 			},
 		},
 		{
-			name: "ShouldSucceed_WhenPolicyHasNoRequirements",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{{
-					Priority:                1,
-					PriorityName:            "High",
-					IntraFlowDispatchPolicy: intra.RegisteredPolicyName("policy-without-req"),
-				}},
+			name: "ShouldRespectGlobalOverrides",
+			opts: []ConfigOption{
+				WithInitialShardCount(10),
+				WithMaxBytes(5000),
+				WithFlowGCTimeout(1 * time.Hour),
+				WithPriorityBand(mustBand(t, 1, "High")),
 			},
-			opts: []configOption{
-				withIntraFlowDispatchPolicyFactory(
-					func(_ intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
-						return &mocks.MockIntraFlowDispatchPolicy{
-							NameV:                      "policy-without-req",
-							RequiredQueueCapabilitiesV: []framework.QueueCapability{},
-						}, nil
-					}),
+			assertion: func(t *testing.T, cfg *Config) {
+				assert.Equal(t, 10, cfg.InitialShardCount)
+				assert.Equal(t, uint64(5000), cfg.MaxBytes)
+				assert.Equal(t, 1*time.Hour, cfg.FlowGCTimeout)
 			},
-			expectErr: false,
 		},
 		{
-			name: "ShouldSucceed_WhenPolicyAndQueueAreCompatible",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{{
-					Priority:                1,
-					PriorityName:            "High",
-					IntraFlowDispatchPolicy: intra.RegisteredPolicyName("policy-with-reqs"),
-					Queue:                   queue.RegisteredQueueName("queue-with-reqs-capability"),
-				}},
+			name: "ShouldApplyBandDefaults_WithRawStructLiterals",
+			opts: []ConfigOption{
+				// Simulate a user passing a manually constructed struct (bypassing NewPriorityBandConfig).
+				WithPriorityBand(&PriorityBandConfig{Priority: 1, PriorityName: "Raw"}),
 			},
-			opts: []configOption{
-				withIntraFlowDispatchPolicyFactory(
-					func(_ intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
-						return &mocks.MockIntraFlowDispatchPolicy{
-							RequiredQueueCapabilitiesV: []framework.QueueCapability{"capability-A", "capability-B"},
-						}, nil
-					}),
-				withQueueFactory(
-					func(_ queue.RegisteredQueueName, _ framework.ItemComparator) (framework.SafeQueue, error) {
-						return &mocks.MockSafeQueue{
-							CapabilitiesV: []framework.QueueCapability{"capability-A", "capability-B", "capability-C"},
-						}, nil
-					}),
+			assertion: func(t *testing.T, cfg *Config) {
+				require.Contains(t, cfg.PriorityBands, 1)
+				band := cfg.PriorityBands[1]
+				assert.Equal(t, defaultQueue, band.Queue, "Queue should be defaulted even for raw struct inputs")
+				assert.Equal(t, defaultIntraFlowDispatchPolicy, band.IntraFlowDispatchPolicy)
 			},
-			expectErr: false,
 		},
-		// --- Core Behavioral Contracts ---
+
+		// --- Validation Errors (Global) ---
 		{
-			name: "ShouldReturnDeepCopy_AndNotMutateInput",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{{Priority: 1, PriorityName: "High"}},
-			},
-			assertion: func(t *testing.T, originalCfg Config, newCfg *Config) {
-				newCfg.PriorityBands[0].PriorityName = "changed"
-				assert.NotEqual(t, originalCfg.PriorityBands[0].PriorityName, newCfg.PriorityBands[0].PriorityName,
-					"Original config should not be mutated by changes to the new config")
-			},
+			name:          "ShouldError_WhenNoPriorityBandsDefined",
+			opts:          []ConfigOption{WithInitialShardCount(1)},
+			expectErr:     true,
+			expectedErrIs: nil, // Generic error expected.
 		},
-		// --- Input Validation Errors ---
 		{
-			name:      "ShouldError_WhenNoPriorityBandsAreDefined",
-			input:     Config{PriorityBands: []PriorityBandConfig{}},
+			name:      "ShouldError_WhenInitialShardCountIsInvalid",
+			opts:      []ConfigOption{WithInitialShardCount(0)}, // Option itself should return error.
 			expectErr: true,
 		},
 		{
-			name:      "ShouldError_WhenPriorityNameIsMissing",
-			input:     Config{PriorityBands: []PriorityBandConfig{{Priority: 1}}},
+			name:      "ShouldError_WhenFlowGCTimeoutIsInvalid",
+			opts:      []ConfigOption{WithFlowGCTimeout(-1 * time.Second)},
 			expectErr: true,
 		},
+
+		// --- Validation Errors (Bands) ---
 		{
-			name: "ShouldError_WhenPriorityLevelsAreDuplicated",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{
-					{Priority: 1, PriorityName: "High"},
-					{Priority: 1, PriorityName: "Also High"},
-				},
+			name: "ShouldError_WhenDuplicatePriorityLevelAdded",
+			opts: []ConfigOption{
+				WithPriorityBand(mustBand(t, 1, "A")),
+				WithPriorityBand(mustBand(t, 1, "B")), // Same priority level
 			},
 			expectErr: true,
 		},
 		{
-			name: "ShouldError_WhenPriorityNamesAreDuplicated",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{
-					{Priority: 1, PriorityName: "High"},
-					{Priority: 2, PriorityName: "High"},
-				},
+			name: "ShouldError_WhenDuplicatePriorityNameAdded",
+			opts: []ConfigOption{
+				WithPriorityBand(mustBand(t, 1, "High")),
+				WithPriorityBand(mustBand(t, 2, "High")), // Same name
 			},
 			expectErr: true,
 		},
-		// --- Plugin and Compatibility Errors ---
 		{
-			name: "ShouldError_WhenPolicyAndQueueAreIncompatible",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{{
-					Priority:                1,
-					PriorityName:            "High",
-					IntraFlowDispatchPolicy: intra.RegisteredPolicyName("policy-with-req"),
-					Queue:                   queue.ListQueueName,
-				}},
-			},
-			opts: []configOption{withIntraFlowDispatchPolicyFactory(
-				func(_ intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
-					return &mocks.MockIntraFlowDispatchPolicy{
-						RequiredQueueCapabilitiesV: []framework.QueueCapability{framework.QueueCapability("required-capability")},
-					}, nil
-				})},
-			expectErr:     true,
-			expectedErrIs: contracts.ErrPolicyQueueIncompatible,
-		},
-		{
-			name: "ShouldError_WhenQueueFactoryFails",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{{
-					Priority:                1,
-					PriorityName:            "High",
-					Queue:                   queue.RegisteredQueueName("failing-queue"),
-					IntraFlowDispatchPolicy: intra.RegisteredPolicyName("policy-with-req"),
-				}},
+			name: "ShouldError_WhenBandIsNil",
+			opts: []ConfigOption{
+				WithPriorityBand(nil),
 			},
 			expectErr: true,
-			opts: []configOption{
-				withIntraFlowDispatchPolicyFactory( // Forces queue instance creation for validating capabilities.
-					func(name intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
-						return &mocks.MockIntraFlowDispatchPolicy{
-							NameV:                      string(name),
-							RequiredQueueCapabilitiesV: []framework.QueueCapability{"required-capability"},
-						}, nil
-					},
-				),
-				withQueueFactory(func(_ queue.RegisteredQueueName, _ framework.ItemComparator) (framework.SafeQueue, error) {
-					return nil, errors.New("queue creation failed")
-				}),
-			},
 		},
 		{
-			name: "ShouldError_WhenPolicyFactoryFails",
-			input: Config{
-				PriorityBands: []PriorityBandConfig{{
-					Priority:                1,
-					PriorityName:            "High",
-					IntraFlowDispatchPolicy: intra.RegisteredPolicyName("failing-policy"),
-				}},
+			name: "ShouldError_WhenBandNameMissing",
+			opts: []ConfigOption{
+				// Use raw struct to bypass NewPriorityBandConfig validation.
+				WithPriorityBand(&PriorityBandConfig{Priority: 1}),
 			},
-			opts: []configOption{withIntraFlowDispatchPolicyFactory(
-				func(_ intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
-					return nil, errors.New("policy creation failed")
-				})},
 			expectErr: true,
 		},
-		// --- Defaulting of Invalid Values ---
+
+		// --- Compatibility Checks ---
 		{
-			name: "ShouldApplyDefault_WhenInitialShardCountIsInvalid",
-			input: Config{
-				InitialShardCount: -1,
-				PriorityBands:     []PriorityBandConfig{{Priority: 1, PriorityName: "High"}},
-			},
-			assertion: func(t *testing.T, _ Config, newCfg *Config) {
-				assert.Equal(t, defaultInitialShardCount, newCfg.InitialShardCount,
-					"Invalid InitialShardCount should be corrected to the default")
+			name: "ShouldError_WhenCapabilityCheckerFails",
+			opts: []ConfigOption{
+				WithPriorityBand(mustBand(t, 1, "High")),
+				withCapabilityChecker(&mockCapabilityChecker{
+					checkCompatibilityFunc: func(p intra.RegisteredPolicyName, q queue.RegisteredQueueName) error {
+						return contracts.ErrPolicyQueueIncompatible
+					},
+				}),
 			},
+			expectErr:     true,
+			expectedErrIs: contracts.ErrPolicyQueueIncompatible,
 		},
 		{
-			name: "ShouldApplyDefault_WhenFlowGCTimeoutIsInvalid",
-			input: Config{
-				FlowGCTimeout: 0,
-				PriorityBands: []PriorityBandConfig{{Priority: 1, PriorityName: "High"}},
-			},
-			assertion: func(t *testing.T, _ Config, newCfg *Config) {
-				assert.Equal(t, defaultFlowGCTimeout, newCfg.FlowGCTimeout,
-					"Invalid FlowGCTimeout should be corrected to the default")
+			name: "ShouldError_WhenDefaultRuntimeCheckerDetectsUnknownPolicy",
+			opts: []ConfigOption{
+				WithPriorityBand(mustBand(t, 1, "BadBand", WithIntraFlowPolicy("non-existent-policy"))),
 			},
+			expectErr: true,
 		},
 		{
-			name: "ShouldApplyDefault_WhenEventChannelBufferSizeIsInvalid",
-			input: Config{
-				EventChannelBufferSize: -1,
-				PriorityBands:          []PriorityBandConfig{{Priority: 1, PriorityName: "High"}},
-			},
-			assertion: func(t *testing.T, _ Config, newCfg *Config) {
-				assert.Equal(t, defaultEventChannelBufferSize, newCfg.EventChannelBufferSize,
-					"Invalid EventChannelBufferSize should be corrected to the default")
+			name: "ShouldError_WhenDefaultRuntimeCheckerDetectsUnknownQueue",
+			opts: []ConfigOption{
+				WithPriorityBand(mustBand(t, 1, "BadBand",
+					WithIntraFlowPolicy(fcfs.FCFSPolicyName),
+					WithQueue("non-existent-queue"),
+				)),
 			},
+			expectErr: true,
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			originalInput := tc.input.deepCopy()
-			validatedCfg, err := newConfig(tc.input, tc.opts...)
+
+			cfg, err := NewConfig(tc.opts...)
 
 			if tc.expectErr {
-				require.Error(t, err, "expected an error but got nil")
+				require.Error(t, err, "expected validation error")
 				if tc.expectedErrIs != nil {
-					assert.ErrorIs(t, err, tc.expectedErrIs, "error should wrap the expected error type")
+					assert.ErrorIs(t, err, tc.expectedErrIs)
 				}
-				assert.Nil(t, validatedCfg, "validatedCfg should be nil on error")
+				assert.Nil(t, cfg, "config should be nil on error")
 			} else {
-				require.NoError(t, err, "expected no error but got: %v", err)
-				require.NotNil(t, validatedCfg, "validatedCfg should not be nil on success")
+				require.NoError(t, err, "unexpected configuration error")
+				require.NotNil(t, cfg, "config should not be nil on success")
 				if tc.assertion != nil {
-					tc.assertion(t, *originalInput, validatedCfg)
+					tc.assertion(t, cfg)
 				}
-
-				// Ensure the original config is not mutated.
-				assert.Equal(t, *originalInput, tc.input, "input config should not be mutated")
 			}
 		})
 	}
 }
 
-func TestConfig_Partition(t *testing.T) {
+func TestNewPriorityBandConfig(t *testing.T) {
 	t.Parallel()
-	baseCfg, err := newConfig(Config{
-		MaxBytes: 103, // Will not distribute evenly
-		PriorityBands: []PriorityBandConfig{
-			{Priority: 1, PriorityName: "High", MaxBytes: 55}, // Will not distribute evenly
-			{Priority: 2, PriorityName: "Low", MaxBytes: 0},   // Will be defaulted to 1,000,000,000
-			{Priority: 3, PriorityName: "Mid", MaxBytes: 20},  // Will distribute evenly
-		},
-	})
-	require.NoError(t, err, "Test setup: creating the base config should not fail")
-
-	t.Run("ShouldDistributeWithRemainderCorrectly", func(t *testing.T) {
-		t.Parallel()
-		totalShards := 10
-		// Global: 103 / 10 = 10 rem 3. First 3 shards get 11, rest get 10.
-		// Band 1: 55 / 10 = 5 rem 5. First 5 shards get 6, rest get 5.
-		// Band 2 (defaulted): 1,000,000,000 / 10 = 100,000,000 for all shards.
-		// Band 3: 20 / 10 = 2 rem 0. All shards get 2.
-		expectedGlobalBytes := []uint64{11, 11, 11, 10, 10, 10, 10, 10, 10, 10}
-		expectedBand1Bytes := []uint64{6, 6, 6, 6, 6, 5, 5, 5, 5, 5}
-
-		var totalGlobal, totalBand1, totalBand2, totalBand3 uint64
-		for i := range totalShards {
-			shardCfg := baseCfg.partition(i, totalShards)
-			require.NotNil(t, shardCfg, "Partitioned config for shard %d should not be nil", i)
-			assert.Equal(t, expectedGlobalBytes[i], shardCfg.MaxBytes, "Global MaxBytes for shard %d is incorrect", i)
-			require.Len(t, shardCfg.PriorityBands, 3, "Partitioned config should have 3 bands")
-			assert.Equal(t, expectedBand1Bytes[i], shardCfg.PriorityBands[0].MaxBytes,
-				"Band 1 MaxBytes for shard %d is incorrect", i)
-
-			expectedBand2Bytes := defaultPriorityBandMaxBytes / uint64(totalShards)
-			assert.Equal(t, expectedBand2Bytes, shardCfg.PriorityBands[1].MaxBytes,
-				"Band 2 MaxBytes for shard %d is incorrect", i)
-			assert.Equal(t, uint64(2), shardCfg.PriorityBands[2].MaxBytes,
-				"Band 3 MaxBytes for shard %d is incorrect", i)
-
-			totalGlobal += shardCfg.MaxBytes
-			totalBand1 += shardCfg.PriorityBands[0].MaxBytes
-			totalBand2 += shardCfg.PriorityBands[1].MaxBytes
-			totalBand3 += shardCfg.PriorityBands[2].MaxBytes
-		}
-		assert.Equal(t, baseCfg.MaxBytes, totalGlobal, "Sum of partitioned global MaxBytes should equal original")
-		assert.Equal(t, baseCfg.PriorityBands[0].MaxBytes, totalBand1,
-			"Sum of partitioned band 1 MaxBytes should equal original")
-		assert.Equal(t, defaultPriorityBandMaxBytes, totalBand2,
-			"Sum of partitioned band 2 MaxBytes should equal the default value")
-		assert.Equal(t, baseCfg.PriorityBands[2].MaxBytes, totalBand3,
-			"Sum of partitioned band 3 MaxBytes should equal original")
-	})
 
-	t.Run("ShouldHandleSingleShard", func(t *testing.T) {
+	t.Run("ShouldApplyUserOverrides", func(t *testing.T) {
 		t.Parallel()
-		shardCfg := baseCfg.partition(0, 1)
-		require.NotNil(t, shardCfg, "Partitioned config for single shard should not be nil")
-		assert.Equal(t, baseCfg.MaxBytes, shardCfg.MaxBytes, "Global MaxBytes should be unchanged for a single shard")
-		require.Len(t, shardCfg.PriorityBands, 3, "Partitioned config should have 3 bands")
-		assert.Equal(t, baseCfg.PriorityBands[0].MaxBytes, shardCfg.PriorityBands[0].MaxBytes,
-			"Band 1 MaxBytes should be unchanged")
+		pb, err := NewPriorityBandConfig(1, "Custom",
+			WithQueue(queue.RegisteredQueueName("CustomQueue")),
+			WithBandMaxBytes(999),
+			WithIntraFlowPolicy("CustomPolicy"),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, queue.RegisteredQueueName("CustomQueue"), pb.Queue)
+		assert.Equal(t, uint64(999), pb.MaxBytes)
+		assert.Equal(t, intra.RegisteredPolicyName("CustomPolicy"), pb.IntraFlowDispatchPolicy)
+		assert.Equal(t, defaultInterFlowDispatchPolicy, pb.InterFlowDispatchPolicy) // Unchanged default
 	})
 
-	t.Run("ShouldPanic_WhenTotalShardsIsInvalid", func(t *testing.T) {
+	t.Run("ShouldError_OnInvalidOptions", func(t *testing.T) {
 		t.Parallel()
-		testCases := []struct {
-			name        string
-			totalShards int
-		}{
-			{"ZeroShards", 0},
-			{"NegativeShards", -1},
-		}
-
-		for _, tc := range testCases {
-			t.Run(tc.name, func(t *testing.T) {
-				t.Parallel()
-				assert.PanicsWithValue(t,
-					fmt.Sprintf("invariant violation: cannot partition into %d partitions", tc.totalShards),
-					func() { baseCfg.partition(0, tc.totalShards) },
-					"Should panic with a specific message when totalShards is not positive",
-				)
-			})
-		}
+		pb, err := NewPriorityBandConfig(1, "Bad", WithQueue(""))
+		assert.Error(t, err, "Should error when setting empty queue")
+		assert.Nil(t, pb)
 	})
 }
 
-func TestConfig_GetBandConfig(t *testing.T) {
+func TestConfig_Partition(t *testing.T) {
 	t.Parallel()
-	cfg, err := newConfig(Config{
-		PriorityBands: []PriorityBandConfig{
-			{Priority: 10, PriorityName: "High"},
-		},
-	})
-	require.NoError(t, err, "Test setup: creating the config should not fail")
 
-	t.Run("ShouldReturnCorrectConfig_ForExistingPriority", func(t *testing.T) {
+	// Setup:
+	// Global: 103 MaxBytes.
+	// Band 1: 55 MaxBytes.
+	// Band 2: 0 MaxBytes (will default to 1GB).
+	// Band 3: 20 MaxBytes.
+	cfg, err := NewConfig(
+		WithMaxBytes(103),
+		WithPriorityBand(mustBand(t, 1, "Band1", WithBandMaxBytes(55))),
+		WithPriorityBand(mustBand(t, 2, "Band2", WithBandMaxBytes(0))), // Explicit 0 implies default behavior via logic.
+		WithPriorityBand(mustBand(t, 3, "Band3", WithBandMaxBytes(20))),
+	)
+	require.NoError(t, err)
+
+	// NewConfig applies defaults. If we passed 0 to NewPriorityBandConfig, it became 1GB.
+	// We need to check what the setup resulted in.
+	expectedBand2Total := defaultPriorityBandMaxBytes
+	assert.Equal(t, expectedBand2Total, cfg.PriorityBands[2].MaxBytes, "Band 2 should have been defaulted")
+
+	t.Run("ShouldDistributeRemainderCorrectly", func(t *testing.T) {
 		t.Parallel()
-		bandCfg, err := cfg.getBandConfig(10)
-		require.NoError(t, err, "Should not error for an existing priority")
-		require.NotNil(t, bandCfg, "Returned band config should not be nil")
-		assert.Equal(t, "High", bandCfg.PriorityName, "Should return the correct band config")
-	})
+		totalShards := 10
 
-	t.Run("ShouldReturnError_ForNonExistentPriority", func(t *testing.T) {
-		t.Parallel()
-		_, err := cfg.getBandConfig(99)
-		assert.ErrorIs(t, err, contracts.ErrPriorityBandNotFound, "Error should wrap ErrPriorityBandNotFound")
+		// Global: 103 / 10 = 10 rem 3. First 3 shards get 11.
+		// Band 1: 55 / 10 = 5 rem 5. First 5 shards get 6.
+		// Band 3: 20 / 10 = 2 rem 0. All get 2.
+
+		var sumGlobal, sumBand1, sumBand2, sumBand3 uint64
+
+		for i := 0; i < totalShards; i++ {
+			shard := cfg.partition(i, totalShards)
+			require.NotNil(t, shard)
+
+			// Accumulate.
+			sumGlobal += shard.MaxBytes
+			sumBand1 += shard.PriorityBands[1].MaxBytes
+			sumBand2 += shard.PriorityBands[2].MaxBytes
+			sumBand3 += shard.PriorityBands[3].MaxBytes
+
+			// Spot check specific shards.
+			if i < 3 {
+				assert.Equal(t, uint64(11), shard.MaxBytes, "Shard %d global bytes mismatch", i)
+			} else {
+				assert.Equal(t, uint64(10), shard.MaxBytes, "Shard %d global bytes mismatch", i)
+			}
+		}
+
+		// Verify totals.
+		assert.Equal(t, cfg.MaxBytes, sumGlobal, "Total global bytes preserved")
+		assert.Equal(t, cfg.PriorityBands[1].MaxBytes, sumBand1, "Total Band 1 bytes preserved")
+		assert.Equal(t, cfg.PriorityBands[2].MaxBytes, sumBand2, "Total Band 2 bytes preserved")
+		assert.Equal(t, cfg.PriorityBands[3].MaxBytes, sumBand3, "Total Band 3 bytes preserved")
 	})
 }
 
-func TestConfig_DeepCopy(t *testing.T) {
+func TestConfig_Clone(t *testing.T) {
 	t.Parallel()
 
-	baseCfg := Config{
-		MaxBytes:               1000,
-		InitialShardCount:      5,
-		FlowGCTimeout:          10 * time.Minute,
-		EventChannelBufferSize: 2048,
-		PriorityBands: []PriorityBandConfig{
-			{Priority: 1, PriorityName: "A"},
-			{Priority: 2, PriorityName: "B"},
-		},
-	}
-	// Create a fully initialized "original" config to be the source of the copy.
-	original, err := newConfig(baseCfg)
-	require.NoError(t, err, "Setup for deep copy should not fail")
+	original, err := NewConfig(
+		WithMaxBytes(1000),
+		WithPriorityBand(mustBand(t, 1, "A")),
+		WithPriorityBand(mustBand(t, 2, "B")),
+	)
+	require.NoError(t, err, "Setup failed")
 
 	t.Run("ShouldReturnNil_ForNilReceiver", func(t *testing.T) {
-		t.Parallel()
 		var nilConfig *Config
-		assert.Nil(t, nilConfig.deepCopy(), "Deep copy of a nil config should be nil")
+		assert.Nil(t, nilConfig.Clone())
 	})
 
-	t.Run("ShouldCreateIdenticalButSeparateObject", func(t *testing.T) {
-		t.Parallel()
-		clone := original.deepCopy()
-
-		// Assert that the clone is not the same object in memory.
-		require.NotSame(t, original, clone, "Clone should be a new object in memory")
-		require.NotSame(t, &original.PriorityBands, &clone.PriorityBands, "PriorityBands slice should be a new object")
-		require.NotSame(t, &original.priorityBandMap, &clone.priorityBandMap, "priorityBandMap should be a new object")
-		assert.Equal(t, original.MaxBytes, clone.MaxBytes, "MaxBytes should be copied")
-		assert.Equal(t, original.InitialShardCount, clone.InitialShardCount, "InitialShardCount should be copied")
-		assert.Equal(t, original.FlowGCTimeout, clone.FlowGCTimeout, "FlowGCTimeout should be copied")
-		assert.Equal(t, original.EventChannelBufferSize, clone.EventChannelBufferSize,
-			"EventChannelBufferSize should be copied")
-		assert.Equal(t, original.PriorityBands, clone.PriorityBands, "PriorityBands should be deeply copied")
-	})
+	t.Run("ShouldCreateDeepCopy", func(t *testing.T) {
+		clone := original.Clone()
 
-	t.Run("ShouldBeIndependent_AfterCopying", func(t *testing.T) {
-		t.Parallel()
-		clone := original.deepCopy()
-
-		// Modify the clone
-		clone.MaxBytes = 2000
-		clone.PriorityBands[0].PriorityName = "modified"
-		// This modification through the map should also modify the slice element, as the map holds pointers.
-		clone.priorityBandMap[clone.PriorityBands[0].Priority].PriorityName = "modified-in-map"
-		clone.queueFactory = func(_ queue.RegisteredQueueName, _ framework.ItemComparator) (framework.SafeQueue, error) {
-			return nil, nil
-		}
-
-		// Assert that the original remains unchanged by comparing field by field for clarity.
-		assert.Equal(t, uint64(1000), original.MaxBytes, "Original MaxBytes should not be changed")
-		assert.Equal(t, 5, original.InitialShardCount, "Original InitialShardCount should not be changed")
-		assert.Equal(t, "A", original.PriorityBands[0].PriorityName, "Original PriorityBands slice should not be changed")
-		assert.Equal(t, "A", original.priorityBandMap[1].PriorityName, "Original priorityBandMap should not be changed")
-		assert.NotNil(t, original.queueFactory, "Original queueFactory should not be nil")
-
-		// Verify the change was contained to the clone. The final value should be the one set via the map.
-		assert.Equal(t, "modified-in-map", clone.PriorityBands[0].PriorityName,
-			"Clone's slice should reflect the change from the map")
-		assert.Equal(t, "modified-in-map", clone.priorityBandMap[1].PriorityName, "Clone's map should reflect the change")
+		require.NotSame(t, original, clone, "Struct pointers should differ")
+		require.NotSame(t, original.PriorityBands[1], clone.PriorityBands[1],
+			"Map values (pointers to bands) should differ")
+		assert.Equal(t, original.MaxBytes, clone.MaxBytes)
+		assert.Equal(t, original.PriorityBands[1].PriorityName, clone.PriorityBands[1].PriorityName)
 	})
-}
 
-func TestShardConfig_GetBandConfig(t *testing.T) {
-	t.Parallel()
-	baseCfg, err := newConfig(Config{
-		PriorityBands: []PriorityBandConfig{
-			{Priority: 10, PriorityName: "High"},
-			{Priority: 20, PriorityName: "Low"},
-		},
-	})
-	require.NoError(t, err, "Test setup: creating the base config should not fail")
-	shardCfg := baseCfg.partition(0, 2)
-	require.NotNil(t, shardCfg, "Test setup: partitioned config should not be nil")
+	t.Run("ShouldIsolateModifications", func(t *testing.T) {
+		clone := original.Clone()
 
-	t.Run("ShouldReturnCorrectConfig_ForExistingPriority", func(t *testing.T) {
-		t.Parallel()
-		bandCfg, err := shardCfg.getBandConfig(20)
-		require.NoError(t, err, "Should not error for an existing priority")
-		require.NotNil(t, bandCfg, "Returned band config should not be nil")
-		assert.Equal(t, "Low", bandCfg.PriorityName, "Should return the correct partitioned band config")
-	})
+		// Modify the clone's map entry.
+		clone.PriorityBands[1].MaxBytes = 99999
+		clone.PriorityBands[1].PriorityName = "Modified"
 
-	t.Run("ShouldReturnError_ForNonExistentPriority", func(t *testing.T) {
-		t.Parallel()
-		_, err := shardCfg.getBandConfig(99)
-		assert.ErrorIs(t, err, contracts.ErrPriorityBandNotFound, "Error should wrap ErrPriorityBandNotFound")
+		assert.Equal(t, defaultPriorityBandMaxBytes, original.PriorityBands[1].MaxBytes)
+		assert.Equal(t, "A", original.PriorityBands[1].PriorityName)
+		assert.Equal(t, uint64(99999), clone.PriorityBands[1].MaxBytes)
+		assert.Equal(t, "Modified", clone.PriorityBands[1].PriorityName)
 	})
 }
diff --git a/pkg/epp/flowcontrol/registry/registry_test.go b/pkg/epp/flowcontrol/registry/registry_test.go
index 494bce0..4d26a7b 100644
--- a/pkg/epp/flowcontrol/registry/registry_test.go
+++ b/pkg/epp/flowcontrol/registry/registry_test.go
@@ -18,7 +18,6 @@ package registry
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sync"
 	"testing"
@@ -30,9 +29,8 @@ import (
 	testclock "k8s.io/utils/clock/testing"
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/interflow"
 	intra "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types/mocks"
 )
@@ -57,28 +55,37 @@ type harnessOptions struct {
 func newRegistryTestHarness(t *testing.T, opts harnessOptions) *registryTestHarness {
 	t.Helper()
 
-	var config Config
+	var cfg *Config
+	var err error
+
 	if opts.config != nil {
-		config = *opts.config
+		cfg = opts.config.Clone()
+		if opts.initialShardCount > 0 {
+			cfg.InitialShardCount = opts.initialShardCount
+		}
 	} else {
-		config = Config{
-			FlowGCTimeout: 5 * time.Minute, // Use a realistic but controllable GC time.
-			PriorityBands: []PriorityBandConfig{
-				{Priority: highPriority, PriorityName: "High"},
-				{Priority: lowPriority, PriorityName: "Low"},
-			},
+		shardCount := 1
+		if opts.initialShardCount > 0 {
+			shardCount = opts.initialShardCount
 		}
-	}
-	if opts.initialShardCount > 0 {
-		config.InitialShardCount = opts.initialShardCount
-	}
 
-	validatedCfg, err := config.ValidateAndApplyDefaults()
-	require.NoError(t, err, "Test setup: validating config should not fail")
+		highBand, err := NewPriorityBandConfig(highPriority, "High")
+		require.NoError(t, err)
+		lowBand, err := NewPriorityBandConfig(lowPriority, "Low")
+		require.NoError(t, err)
+
+		cfg, err = NewConfig(
+			WithInitialShardCount(shardCount),
+			WithFlowGCTimeout(5*time.Minute),
+			WithPriorityBand(highBand),
+			WithPriorityBand(lowBand),
+		)
+		require.NoError(t, err, "Test setup: failed to create default config")
+	}
 
 	fakeClock := testclock.NewFakeClock(time.Now())
 	registryOpts := []RegistryOption{withClock(fakeClock)}
-	fr, err := NewFlowRegistry(*validatedCfg, logr.Discard(), registryOpts...)
+	fr, err := NewFlowRegistry(cfg, logr.Discard(), registryOpts...)
 	require.NoError(t, err, "Test setup: NewFlowRegistry should not fail")
 
 	// Start the GC loop in the background.
@@ -137,17 +144,18 @@ func TestFlowRegistry_New(t *testing.T) {
 
 	t.Run("ShouldFail_WhenInitialShardCreationFails", func(t *testing.T) {
 		t.Parallel()
-		config, err := newConfig(
-			Config{PriorityBands: []PriorityBandConfig{{Priority: highPriority, PriorityName: "A"}}},
-			withInterFlowDispatchPolicyFactory(func(interflow.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error) {
-				return nil, errors.New("injected factory failure")
-			}),
-		)
+
+		badBand, err := NewPriorityBandConfig(highPriority, "A", WithInterFlowPolicy("non-existent-policy"))
+		require.NoError(t, err)
+
+		config, err := NewConfig(WithPriorityBand(badBand))
 		require.NoError(t, err, "Test setup: creating the config object itself should not fail")
-		_, err = NewFlowRegistry(*config, logr.Discard())
+
+		_, err = NewFlowRegistry(config, logr.Discard())
+
 		require.Error(t, err, "NewFlowRegistry should fail when initial shard setup fails")
-		assert.Contains(t, err.Error(), "injected factory failure",
-			"Error message should reflect the root cause from the failing plugin factory")
+		assert.Contains(t, err.Error(), "failed to create inter-flow policy",
+			"Error message should reflect the root cause (policy creation failure)")
 	})
 }
 
@@ -189,19 +197,35 @@ func TestFlowRegistry_WithConnection_AndHandle(t *testing.T) {
 
 	t.Run("ShouldFail_WhenJITFails", func(t *testing.T) {
 		t.Parallel()
-		h := newRegistryTestHarness(t, harnessOptions{})
-		h.fr.config.intraFlowDispatchPolicyFactory = func(intra.RegisteredPolicyName) (framework.IntraFlowDispatchPolicy, error) {
-			return nil, errors.New("injected factory failure")
-		}
+
+		badPolicyName := intra.RegisteredPolicyName("non-existent-policy")
+		badBand, err := NewPriorityBandConfig(highPriority, "High", WithIntraFlowPolicy(badPolicyName))
+		require.NoError(t, err)
+
+		// Create a Config that uses a mock checker to bypass the strict validation.
+		// The default checker would reject "non-existent-policy", but our mock says it's fine.
+		// This allows us to instantiate the Registry with a latent configuration bomb.
+		cfg, err := NewConfig(
+			WithPriorityBand(badBand),
+			withCapabilityChecker(&mockCapabilityChecker{
+				checkCompatibilityFunc: func(_ intra.RegisteredPolicyName, _ queue.RegisteredQueueName) error {
+					return nil // Approve everything.
+				},
+			}),
+		)
+		require.NoError(t, err)
+
+		h := newRegistryTestHarness(t, harnessOptions{config: cfg})
 		key := types.FlowKey{ID: "test-flow", Priority: highPriority}
 
-		err := h.fr.WithConnection(key, func(conn contracts.ActiveFlowConnection) error {
-			t.Fatal("Callback must not be executed when the flow failes to register JIT")
+		err = h.fr.WithConnection(key, func(conn contracts.ActiveFlowConnection) error {
+			t.Fatal("Callback must not be executed when the flow fails to register JIT")
 			return nil
 		})
 
 		require.Error(t, err, "WithConnection must return an error for a failed flow JIT registration")
-		assert.ErrorContains(t, err, "injected factory failure", "The returned error must propagate the reason")
+		assert.ErrorContains(t, err, "no IntraFlowDispatchPolicy registered",
+			"The returned error must propagate the reason")
 	})
 
 	t.Run("Handle_Shards_ShouldReturnAllActiveShardsAndBeACopy", func(t *testing.T) {
@@ -518,19 +542,22 @@ func TestFlowRegistry_UpdateShardCount(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			config := Config{
-				MaxBytes: globalCapacity,
-				PriorityBands: []PriorityBandConfig{
-					{Priority: highPriority, PriorityName: "A", MaxBytes: bandCapacity},
-				},
-				InitialShardCount: tc.initialShardCount,
-			}
 
-			h := newRegistryTestHarness(t, harnessOptions{config: &config})
+			band, err := NewPriorityBandConfig(highPriority, "A", WithBandMaxBytes(bandCapacity))
+			require.NoError(t, err)
+
+			config, err := NewConfig(
+				WithMaxBytes(globalCapacity),
+				WithInitialShardCount(tc.initialShardCount),
+				WithPriorityBand(band),
+			)
+			require.NoError(t, err, "Test setup: creating config should not fail")
+
+			h := newRegistryTestHarness(t, harnessOptions{config: config})
 			key := types.FlowKey{ID: "flow", Priority: highPriority}
 			h.openConnectionOnFlow(key)
 
-			err := h.fr.updateShardCount(tc.targetShardCount)
+			err = h.fr.updateShardCount(tc.targetShardCount)
 			if tc.expectErrIs != nil {
 				require.Error(t, err, "UpdateShardCount should have returned an error")
 				assert.ErrorIs(t, err, tc.expectErrIs, "Error should be the expected type")
diff --git a/pkg/epp/flowcontrol/registry/shard_test.go b/pkg/epp/flowcontrol/registry/shard_test.go
index c58b71c..28f2bff 100644
--- a/pkg/epp/flowcontrol/registry/shard_test.go
+++ b/pkg/epp/flowcontrol/registry/shard_test.go
@@ -17,7 +17,6 @@ limitations under the License.
 package registry
 
 import (
-	"errors"
 	"fmt"
 	"sync"
 	"testing"
@@ -28,7 +27,6 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/interflow"
 	intra "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/policies/intraflow/dispatch"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework/plugins/queue"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
@@ -59,12 +57,11 @@ type shardTestHarness struct {
 // newShardTestHarness initializes a `shardTestHarness` with a default configuration.
 func newShardTestHarness(t *testing.T) *shardTestHarness {
 	t.Helper()
-	globalConfig, err := newConfig(Config{
-		PriorityBands: []PriorityBandConfig{
-			{Priority: highPriority, PriorityName: "High"},
-			{Priority: lowPriority, PriorityName: "Low"},
-		},
-	})
+
+	globalConfig, err := NewConfig(
+		WithPriorityBand(&PriorityBandConfig{Priority: highPriority, PriorityName: "High"}),
+		WithPriorityBand(&PriorityBandConfig{Priority: lowPriority, PriorityName: "Low"}),
+	)
 	require.NoError(t, err, "Test setup: validating and defaulting config should not fail")
 
 	statsPropagator := &mockStatsPropagator{}
@@ -73,7 +70,6 @@ func newShardTestHarness(t *testing.T) *shardTestHarness {
 		"test-shard-1",
 		shardConfig, logr.Discard(),
 		statsPropagator.propagate,
-		interflow.NewPolicyFromName,
 	)
 	require.NoError(t, err, "Test setup: newShard should not return an error with valid configuration")
 
@@ -146,14 +142,21 @@ func TestShard_New(t *testing.T) {
 
 	t.Run("ShouldFail_WhenInterFlowPolicyFactoryFails", func(t *testing.T) {
 		t.Parallel()
-		shardConfig, _ := newConfig(Config{PriorityBands: []PriorityBandConfig{
-			{Priority: highPriority, PriorityName: "High"},
-		}})
-		failingFactory := func(interflow.RegisteredPolicyName) (framework.InterFlowDispatchPolicy, error) {
-			return nil, errors.New("policy not found")
-		}
-		_, err := newShard("test-shard-1", shardConfig.partition(0, 1), logr.Discard(), nil, failingFactory)
-		require.Error(t, err, "newShard must fail if the inter-flow policy cannot be instantiated during initialization")
+
+		badConfig, err := NewConfig(
+			WithPriorityBand(&PriorityBandConfig{
+				Priority:                highPriority,
+				PriorityName:            "High",
+				InterFlowDispatchPolicy: "non-existent-policy",
+			}),
+		)
+		require.NoError(t, err, "Config validation currently validates IntraFlow policies, but not InterFlow policies")
+
+		_, err = newShard("test-shard-1", badConfig.partition(0, 1), logr.Discard(), nil)
+
+		require.Error(t, err, "newShard must fail if the inter-flow policy cannot be instantiated")
+		assert.Contains(t, err.Error(), "failed to create inter-flow policy",
+			"Error message should indicate policy creation failure")
 	})
 }