fix(gateway): set jwksUrl so the dispatcher builds a validator

cmd/gateway/main.go:144 constructs the adapter validator only when both cfg.Auth.Enabled and cfg.Auth.JWKSURL are set. The config left jwksUrl unset on purpose, so jwtValidator stayed nil and every adapter request passed with one warning line: auth required but jwtValidator nil; request unauthenticated That made auth.required cosmetic: sqs, workflow, memory, s3 and iam all answered unauthenticated calls. Set the JWKS URL to the ZITADEL key endpoint.
feat(gateway): require a token on every adapter
2026-10-27 01:17:09 -07:00 · 2026-10-27 01:11:18 -07:00 · 2026-10-27 01:11:18 -07:00 · 2026-09-30 12:19:24 +00:00 · 2026-09-30 10:56:10 +00:00 · 2026-09-29 15:46:39 +00:00
23 changed files with 2045 additions and 241 deletions
@@ -4,9 +4,13 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"math"
 	"net/http"
 	"os"
 	"os/signal"
+	"runtime"
+	"strconv"
+	"strings"
 	"syscall"

 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
@@ -21,6 +25,10 @@ import (
 )

 func main() {
+	// Size the Go scheduler to the container's CPU quota before any server
+	// goroutine starts.
+	configureGOMAXPROCS()
+
 	ctx := context.Background()

 	// Initialize OpenTelemetry tracing
@@ -177,3 +185,53 @@ func main() {
 	log.Printf("gateway shutdown complete")
 	os.Exit(0)
 }
+
+// configureGOMAXPROCS sets GOMAXPROCS from the cgroup CPU quota so the runtime
+// sizes its scheduler to the container limit rather than the host CPU count.
+// The default is left alone when no quota is configured.
+func configureGOMAXPROCS() {
+	quota, period := readCPUQuota()
+	if quota <= 0 || period <= 0 {
+		return
+	}
+	n := int(math.Ceil(float64(quota) / float64(period)))
+	if n < 1 {
+		n = 1
+	}
+	runtime.GOMAXPROCS(n)
+	log.Printf("GOMAXPROCS set to %d (cgroup cpu quota %d/%d)", n, quota, period)
+}
+
+// readCPUQuota returns the CPU quota and period from cgroup v2, falling back to
+// cgroup v1. A quota of 0 means no limit.
+func readCPUQuota() (int64, int64) {
+	// cgroup v2: "<quota> <period>" or "max <period>".
+	if data, err := os.ReadFile("/sys/fs/cgroup/cpu.max"); err == nil {
+		fields := strings.Fields(string(data))
+		if len(fields) >= 2 {
+			period, perr := strconv.ParseInt(fields[1], 10, 64)
+			if perr == nil && period > 0 {
+				if fields[0] == "max" {
+					return 0, period
+				}
+				if quota, qerr := strconv.ParseInt(fields[0], 10, 64); qerr == nil {
+					return quota, period
+				}
+			}
+		}
+	}
+
+	// cgroup v1: cpu.cfs_quota_us is -1 when unlimited.
+	period := int64(100000)
+	if data, err := os.ReadFile("/sys/fs/cgroup/cpu/cpu.cfs_period_us"); err == nil {
+		if v, perr := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64); perr == nil && v > 0 {
+			period = v
+		}
+	}
+	if data, err := os.ReadFile("/sys/fs/cgroup/cpu/cpu.cfs_quota_us"); err == nil {
+		if v, qerr := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64); qerr == nil {
+			return v, period
+		}
+	}
+	return 0, 0
+}
@@ -5,39 +5,51 @@ go 1.26.0
 require (
 	github.com/MicahParks/keyfunc/v2 v2.1.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
+	github.com/google/uuid v1.6.0
+	github.com/zitadel/oidc/v3 v3.49.2
+	github.com/zitadel/zitadel-go/v3 v3.30.0
 	go.opentelemetry.io/otel v1.46.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.46.0
 	go.opentelemetry.io/otel/sdk v1.46.0
 	go.opentelemetry.io/otel/trace v1.46.0
 	go.temporal.io/api v1.63.5
 	go.temporal.io/sdk v1.48.0
+	golang.org/x/net v0.58.0
 	google.golang.org/grpc v1.83.2
 	google.golang.org/protobuf v1.36.12
 	gopkg.in/yaml.v3 v3.0.1
 )

 require (
+	github.com/bmatcuk/doublestar/v4 v4.10.0 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/envoyproxy/protoc-gen-validate v1.3.3 // indirect
 	github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect
+	github.com/go-chi/chi/v5 v5.3.1 // indirect
+	github.com/go-jose/go-jose/v4 v4.1.4 // indirect
 	github.com/go-logr/logr v1.4.4 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/mock v1.6.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
-	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // indirect
+	github.com/gorilla/securecookie v1.1.2 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 // indirect
+	github.com/muhlemmer/gu v0.3.1 // indirect
+	github.com/muhlemmer/httpforwarded v0.1.0 // indirect
 	github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect
 	github.com/nexus-rpc/sdk-go v0.7.0 // indirect
 	github.com/robfig/cron v1.2.0 // indirect
+	github.com/rs/cors v1.11.1 // indirect
 	github.com/stretchr/objx v0.5.3 // indirect
 	github.com/stretchr/testify v1.12.1 // indirect
+	github.com/zitadel/schema v1.3.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.46.0 // indirect
 	go.opentelemetry.io/otel/metric v1.46.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.11.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.5 // indirect
-	golang.org/x/net v0.58.0 // indirect
+	golang.org/x/oauth2 v0.36.0 // indirect
 	golang.org/x/sync v0.22.0 // indirect
 	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.41.0 // indirect
@@ -1,11 +1,19 @@
 github.com/MicahParks/keyfunc/v2 v2.1.0 h1:6ZXKb9Rp6qp1bDbJefnG7cTH8yMN1IC/4nf+GVjO99k=
 github.com/MicahParks/keyfunc/v2 v2.1.0/go.mod h1:rW42fi+xgLJ2FRRXAfNx9ZA8WpD4OeE/yHVMteCkw9k=
+github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs=
+github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
 github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
 github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/envoyproxy/protoc-gen-validate v1.3.3 h1:MVQghNeW+LZcmXe7SY1V36Z+WFMDjpqGAGacLe2T0ds=
+github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0=
 github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a h1:yDWHCSQ40h88yih2JAcL6Ls/kVkSE8GFACTGVnMPruw=
 github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a/go.mod h1:7Ga40egUymuWXxAe151lTNnCv97MddSOVsjpPPkityA=
+github.com/go-chi/chi/v5 v5.3.1 h1:3j4HZLGZQ3JpMCrPJF/Jl3mYJfWLKBfNJ6quurUGCf8=
+github.com/go-chi/chi/v5 v5.3.1/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
+github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
+github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8=
 github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
@@ -23,8 +31,12 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/securecookie v1.1.2 h1:YCIWL56dvtr73r6715mJs5ZvhtnY73hBvEF8kXD8ePA=
+github.com/gorilla/securecookie v1.1.2/go.mod h1:NfCASbcHqRSY+3a8tlWJwsQap2VX5pwzwo4h3eOamfo=
 github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 h1:sGm2vDRFUrQJO/Veii4h4zG2vvqG6uWNkBHSTqXOZk0=
 github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2/go.mod h1:wd1YpapPLivG6nQgbf7ZkG1hhSOXDhhn4MLTknx2aAc=
+github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 h1:B+8ClL/kCQkRiU82d9xajRPKYMrB7E0MbtzWVi1K4ns=
+github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3/go.mod h1:NbCUVmiS4foBGBHOYlCT25+YmGpJ32dZPi75pGEUpj4=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 h1:/Tnpcb2E0Pz/tN9s3bfEY2Q8ePCEX9iuS+cneUwncnw=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0/go.mod h1:zOBXOsUaBSjKgmH4OGzV1esUpR3oUSCPYVd2cUBjKYY=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
@@ -33,6 +45,10 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/muhlemmer/gu v0.3.1 h1:7EAqmFrW7n3hETvuAdmFmn4hS8W+z3LgKtrnow+YzNM=
+github.com/muhlemmer/gu v0.3.1/go.mod h1:YHtHR+gxM+bKEIIs7Hmi9sPT3ZDUvTN/i88wQpZkrdM=
+github.com/muhlemmer/httpforwarded v0.1.0 h1:x4DLrzXdliq8mprgUMR0olDvHGkou5BJsK/vWUetyzY=
+github.com/muhlemmer/httpforwarded v0.1.0/go.mod h1:yo9czKedo2pdZhoXe+yDkGVbU0TJ0q9oQ90BVoDEtw0=
 github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsxOvp8vZ8hfP0nHhNI80=
 github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y=
 github.com/nexus-rpc/sdk-go v0.7.0 h1:38NrfY5rLnZAiMMs2ZfCKI/CSDzdfJG+27iAgfA8bUI=
@@ -41,6 +57,8 @@ github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
 github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
+github.com/rs/cors v1.11.1 h1:eU3gRzXLRK57F5rKMGMZURNdIG4EoAmX8k94r9wXWHA=
+github.com/rs/cors v1.11.1/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4=
 github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0=
 github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
@@ -48,6 +66,12 @@ github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUm
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+github.com/zitadel/oidc/v3 v3.49.2 h1:yKvB2Hx6rVWp0vOTk1pEyXAPxJusgIHNmr20AxOohmQ=
+github.com/zitadel/oidc/v3 v3.49.2/go.mod h1:HwoguOGo0eem0RK5Gb+P6Q4aQLVinJ9LhomlVEA57ck=
+github.com/zitadel/schema v1.3.2 h1:gfJvt7dOMfTmxzhscZ9KkapKo3Nei3B6cAxjav+lyjI=
+github.com/zitadel/schema v1.3.2/go.mod h1:IZmdfF9Wu62Zu6tJJTH3UsArevs3Y4smfJIj3L8fzxw=
+github.com/zitadel/zitadel-go/v3 v3.30.0 h1:yk/surQ+atLS/ocGaaJdHD/R6VbveFScihxJhIkGZ3Q=
+github.com/zitadel/zitadel-go/v3 v3.30.0/go.mod h1:4R+aki0JFMwIPsr4CIVLYq5DRTVEAfLbehJpFNhFKQw=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
 go.opentelemetry.io/otel v1.46.0 h1:FHt5/CDyVxi/8IM1CH7VE/rRgq3kLHa2mSTVMO8AWyc=
@@ -87,6 +111,8 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
 golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
+golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
+golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -0,0 +1,21 @@
+package auth
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestValidateBearerToken_JWEGetsClearError(t *testing.T) {
+	ti := newTestIssuer(t)
+	host, port, _ := strings.Cut(strings.TrimPrefix(ti.server.URL, "http://"), ":")
+	v, err := NewZitadelValidator(t.Context(), ti.server.URL, nil, ZitadelConfig{
+		Domain: host, InsecurePort: port, HTTPClient: ti.server.Client(),
+	})
+	if err != nil {
+		t.Fatalf("NewZitadelValidator: %v", err)
+	}
+	_, err = v.ValidateBearerToken("Bearer a.b.c.d.e")
+	if err == nil || !strings.Contains(err.Error(), "encrypted JWE") {
+		t.Fatalf("expected a JWE-specific error, got %v", err)
+	}
+}
@@ -3,37 +3,141 @@ package auth
 import (
 	"context"
 	"fmt"
+	gohttp "net/http"
 	"strings"
 	"sync"
 	"time"

 	"github.com/MicahParks/keyfunc/v2"
 	"github.com/golang-jwt/jwt/v5"
+	"github.com/zitadel/zitadel-go/v3/pkg/authorization/oauth"
 )

-// isValidIssuer checks if issuer is from Authentik (any provider/app).
-// Accepts: https://authentik.riotpiao.com/application/o/{provider}/
-func isValidIssuer(iss string) bool {
-	return strings.Contains(iss, "authentik.riotpiao.com/application/o/") &&
-		strings.HasSuffix(iss, "/")
+// issuerMatches reports whether iss is the configured issuer or a sub-path of
+// it. Zitadel emits exactly "https://zitadel.riotpiao.com"; providers that
+// mount an application path append a suffix (Authentik used
+// "/application/o/<app>/"). A trailing slash on either side is ignored.
+//
+// An empty configured issuer disables the domain check; the JWKS signature
+// check still applies. Production configuration always sets an issuer.
+func issuerMatches(iss, configured string) bool {
+	if configured == "" {
+		return iss != ""
+	}
+	iss = strings.TrimSuffix(iss, "/")
+	configured = strings.TrimSuffix(configured, "/")
+	return iss == configured || strings.HasPrefix(iss, configured+"/")
 }

-// Validator validates JWTs against Authentik JWKS.
-// Supports multi-issuer: any Authentik service account provider is accepted
-// (portfolio-agent, memory-agent, api-gw, etc.) because all share the same
-// JWKS signing key.
+// audienceValues normalises the aud claim. Zitadel access tokens carry an
+// array, older providers carry a single string.
+func audienceValues(claims jwt.MapClaims) ([]string, bool) {
+	switch v := claims["aud"].(type) {
+	case string:
+		return []string{v}, v != ""
+	case []string:
+		return v, len(v) > 0
+	case []interface{}:
+		out := make([]string, 0, len(v))
+		for _, item := range v {
+			if s, ok := item.(string); ok {
+				out = append(out, s)
+			}
+		}
+		return out, len(out) > 0
+	case jwt.ClaimStrings:
+		return []string(v), len(v) > 0
+	}
+	return nil, false
+}
+
+// Validator validates JWTs against the JWKS of an OIDC provider (Zitadel).
+// The issuer is configuration-driven: tokens whose iss claim equals the
+// configured issuer, or lives under it as a sub-path, are accepted.
+//
+// When a ZitadelVerifier is attached (see NewZitadelValidator), validation
+// delegates to the official ZITADEL Go SDK and this type only adapts the
+// result to jwt.MapClaims for the rest of the gateway.
 type Validator struct {
-	issuer   string // Not used for validation (kept for logging); issuer regex check is sufficient
-	audience string // Not used for validation; any audience from valid Authentik issuer is accepted
+	issuer   string // Expected issuer (iss claim); empty disables the domain check
+	audience string // Expected audience; currently logged only (any non-empty aud is accepted)
 	jwksURL  string
 	jwks     *keyfunc.JWKS
 	mu       sync.Mutex
+	zitadel  *ZitadelVerifier // non-nil when backed by the ZITADEL SDK
+}
+
+// NewZitadelValidator builds a Validator backed by the ZITADEL Go SDK.
+// issuer is the expected iss claim, used for logging and for the legacy
+// claim checks if the SDK verifier is ever bypassed. clientIDs is the
+// audience allow-list. Discovery runs here, so a wrong domain fails at
+// startup.
+func NewZitadelValidator(ctx context.Context, issuer string, clientIDs []string, cfg ZitadelConfig) (*Validator, error) {
+	return NewZitadelValidatorWithClient(ctx, issuer, clientIDs, cfg, nil)
+}
+
+// NewZitadelValidatorWithClient is NewZitadelValidator with an injected HTTP
+// client, for tests that serve discovery and JWKS from httptest.
+func NewZitadelValidatorWithClient(ctx context.Context, issuer string, clientIDs []string, cfg ZitadelConfig, httpClient *gohttp.Client) (*Validator, error) {
+	cfg.ClientIDs = clientIDs
+	if cfg.HTTPClient == nil {
+		cfg.HTTPClient = httpClient
+	}
+	sdk, err := NewZitadelVerifier(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return &Validator{
+		issuer:  issuer,
+		jwksURL: "", // unused on the SDK path
+		zitadel: sdk,
+	}, nil
+}
+
+// claimsFromIntrospection converts the SDK's verified claim set into
+// jwt.MapClaims so the rest of the gateway keeps one claim representation.
+// The SDK has already checked the signature, exp, nbf and iss; this only
+// copies values across.
+func claimsFromIntrospection(authCtx *oauth.IntrospectionContext) jwt.MapClaims {
+	claims := jwt.MapClaims{}
+	if authCtx == nil {
+		return claims
+	}
+	for k, v := range authCtx.Claims {
+		claims[k] = v
+	}
+	if authCtx.Issuer != "" {
+		claims["iss"] = authCtx.Issuer
+	}
+	if authCtx.Subject != "" {
+		claims["sub"] = authCtx.Subject
+	}
+	if len(authCtx.Audience) > 0 {
+		aud := make([]interface{}, 0, len(authCtx.Audience))
+		for _, a := range authCtx.Audience {
+			aud = append(aud, a)
+		}
+		claims["aud"] = aud
+	}
+	if authCtx.Expiration != 0 {
+		claims["exp"] = float64(authCtx.Expiration)
+	}
+	if authCtx.NotBefore != 0 {
+		claims["nbf"] = float64(authCtx.NotBefore)
+	}
+	if authCtx.IssuedAt != 0 {
+		claims["iat"] = float64(authCtx.IssuedAt)
+	}
+	if authCtx.ClientID != "" {
+		claims["client_id"] = authCtx.ClientID
+	}
+	return claims
 }

 // NewValidator creates a new JWT validator for a service.
-// issuer and audience params are deprecated (ignored for validation) but kept
-// for backward compatibility. Multi-issuer validation via isValidIssuer() is used instead.
-// JWKS fetching is lazy (deferred until first validation).
+// issuer is the expected iss claim (exact value or prefix). audience is kept
+// for compatibility; only non-emptiness is enforced. JWKS fetching is lazy
+// (deferred until first validation).
 func NewValidator(issuer, audience, jwksURL string) *Validator {
 	return &Validator{
 		issuer:   issuer,   // deprecated param, kept for compat
@@ -87,8 +191,31 @@ func (v *Validator) ValidateBearerToken(authHeader string) (jwt.MapClaims, error
 	}

 	// Ensure JWKS is loaded (lazy)
-	if err := v.ensureJWKS(); err != nil {
-		return nil, err
+	if v.jwks != nil || v.zitadel == nil {
+		if err := v.ensureJWKS(); err != nil {
+			return nil, err
+		}
+	}
+
+	// SDK path: ZITADEL Go SDK performs signature, expiry and issuer checks.
+	if v.zitadel != nil {
+		// ZITADEL issues two access-token shapes. A plain JWS has three
+		// dot-separated segments and can be verified locally. A JWE has five,
+		// is encrypted with the instance key, and only ZITADEL can read it.
+		// Report that case explicitly so the operator changes the client to
+		// OIDC_TOKEN_TYPE_BEARER instead of debugging a signature failure.
+		if strings.Count(tokenString, ".") == 4 {
+			return nil, fmt.Errorf("token is an encrypted JWE (5 segments); " +
+				"set accessTokenType=OIDC_TOKEN_TYPE_BEARER on the ZITADEL client so the gateway can verify it locally")
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		authCtx, err := v.zitadel.CheckAuthorization(ctx, authHeader)
+		if err != nil {
+			return nil, fmt.Errorf("token validation failed: %v", err)
+		}
+		return claimsFromIntrospection(authCtx), nil
 	}

 	// Parse and validate
@@ -120,34 +247,35 @@ func (v *Validator) ValidateBearerToken(authHeader string) (jwt.MapClaims, error
 		}
 	}

-	// Check iss (issuer) - accept any Authentik provider issuer
-	// (portfolio-agent, memory-agent, api-gw, etc.)
-	// All use same signing key so JWKS validation is sufficient
+	// Check iss against the configured issuer (exact match or sub-path).
 	if iss, ok := claims["iss"].(string); !ok {
 		return nil, fmt.Errorf("missing issuer claim")
-	} else if !isValidIssuer(iss) {
-		return nil, fmt.Errorf("invalid issuer: %s", iss)
+	} else if !issuerMatches(iss, v.issuer) {
+		return nil, fmt.Errorf("invalid issuer: %s (want %s)", iss, v.issuer)
 	}

-	// Check aud (audience) - accept any Authentik-provided audience
-	// since all Authentik service accounts use the same signing key.
-	// The issuer check above is sufficient to ensure JWT came from Authentik.
-	if aud, ok := claims["aud"].(string); !ok {
+	// Check aud - Zitadel sends an array; accept a string or array.
+	if auds, ok := audienceValues(claims); !ok {
 		return nil, fmt.Errorf("missing audience claim")
-	} else if aud == "" {
+	} else if len(auds) == 0 {
 		return nil, fmt.Errorf("empty audience claim")
 	}
-	// Note: Not hardcoding expected audience. Any audience from a valid Authentik
-	// issuer is accepted, since all service accounts are under the same trust boundary.
+	// The signature check plus the issuer check establish the trust boundary;
+	// the audience value itself is not hardcoded.

 	return claims, nil
 }

 // CheckPermissions checks if claims contain required permission(s).
-// Checks both "permissions" claim (for users) and "roles" claim (for service accounts).
+// Understands three token shapes:
+//   - "permissions": []string            (generic / legacy tokens)
+//   - "roles": []string or map[string]…  (legacy service-account tokens)
+//   - "urn:zitadel:iam:org:project:roles": map[string]…  (Zitadel access tokens,
+//     where the map keys are role keys)
+//
 // Returns true if any required permission is found or wildcard "*" exists.
 func (v *Validator) CheckPermissions(claims jwt.MapClaims, required ...string) bool {
-	// Try permissions claim first (for user tokens)
+	// 1. permissions claim (array of strings)
 	if permsIface, ok := claims["permissions"]; ok {
 		if perms, ok := permsIface.([]interface{}); ok {
 			if v.checkPermList(perms, required...) {
@@ -156,15 +284,45 @@ func (v *Validator) CheckPermissions(claims jwt.MapClaims, required ...string) b
 		}
 	}

-	// Fall back to roles claim (for service account tokens)
+	// 2. roles claim: array (legacy) or map keyed by role name
 	if rolesIface, ok := claims["roles"]; ok {
-		if roles, ok := rolesIface.([]interface{}); ok {
+		switch roles := rolesIface.(type) {
+		case []interface{}:
 			if v.checkPermList(roles, required...) {
 				return true
 			}
+		case map[string]interface{}:
+			if v.checkPermMap(roles, required...) {
+				return true
+			}
 		}
 	}

+	// 3. Zitadel access-token role assertion: map keyed by role key
+	if rolesIface, ok := claims["urn:zitadel:iam:org:project:roles"]; ok {
+		if roles, ok := rolesIface.(map[string]interface{}); ok {
+			if v.checkPermMap(roles, required...) {
+				return true
+			}
+		}
+	}
+
+	return false
+}
+
+// checkPermMap reports whether any key in a role map matches a required
+// permission, or the map carries a wildcard key.
+func (v *Validator) checkPermMap(roles map[string]interface{}, required ...string) bool {
+	for roleKey := range roles {
+		if roleKey == "*" {
+			return true
+		}
+		for _, req := range required {
+			if roleKey == req {
+				return true
+			}
+		}
+	}
 	return false
 }

@@ -0,0 +1,157 @@
+package auth
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	gohttp "net/http"
+	"slices"
+	"strings"
+
+	"github.com/zitadel/oidc/v3/pkg/client"
+	"github.com/zitadel/oidc/v3/pkg/client/rp"
+	"github.com/zitadel/oidc/v3/pkg/oidc"
+	"github.com/zitadel/oidc/v3/pkg/op"
+	"github.com/zitadel/zitadel-go/v3/pkg/authorization/oauth"
+	"github.com/zitadel/zitadel-go/v3/pkg/zitadel"
+)
+
+// ErrAudienceNotAllowed reports a token whose aud claim names none of the
+// configured resource-server client IDs.
+var ErrAudienceNotAllowed = errors.New("audience not allowed")
+
+// ZitadelConfig configures the ZITADEL-backed verifier.
+type ZitadelConfig struct {
+	// Domain is the ZITADEL instance host, for example "zitadel.riotpiao.com".
+	Domain string
+	// InsecurePort, when non-empty, speaks plain HTTP to that port instead of
+	// HTTPS on 443. Use only for a local port-forward.
+	InsecurePort string
+	// ClientIDs is the audience allow-list: a token is accepted when its aud
+	// claim contains at least one of these client IDs. Empty accepts any
+	// non-empty audience, which trusts every client of the same instance.
+	ClientIDs []string
+	// HTTPClient overrides the client used for discovery and JWKS fetches.
+	HTTPClient *gohttp.Client
+}
+
+// ZitadelVerifier validates ZITADEL access tokens with the official
+// github.com/zitadel/zitadel-go SDK.
+//
+// It reuses the SDK's building blocks rather than reimplementing JWT parsing:
+//   - zitadel.New + Origin to address the instance
+//   - client.Discover for OIDC discovery (issuer and JWKS URI)
+//   - rp.NewRemoteKeySet for cached, auto-refreshing signing keys
+//   - op.NewAccessTokenVerifier / op.VerifyAccessToken for signature, expiry
+//     and issuer checks
+//   - oauth.IntrospectionContext for the verified claim set and its
+//     IsGrantedRole helper, which already understands the
+//     urn:zitadel:iam:org:project:roles claim
+//
+// The only behaviour added on top of the SDK is a multi-value audience
+// allow-list, because this gateway is a shared resource server: callers hold
+// their own client IDs (paperless, poimen, api-gw, ...) and the SDK's
+// oauth.WithJWT helper accepts exactly one.
+type ZitadelVerifier struct {
+	instance  *zitadel.Zitadel
+	verifier  *op.AccessTokenVerifier
+	clientIDs []string
+}
+
+// NewZitadelVerifier performs OIDC discovery and returns a ready verifier.
+// Discovery happens once at startup so a misconfigured domain fails fast
+// instead of on the first request.
+func NewZitadelVerifier(ctx context.Context, cfg ZitadelConfig) (*ZitadelVerifier, error) {
+	if cfg.Domain == "" {
+		return nil, errors.New("zitadel: domain is required")
+	}
+
+	httpClient := cfg.HTTPClient
+	if httpClient == nil {
+		httpClient = gohttp.DefaultClient
+	}
+
+	var opts []zitadel.Option
+	if cfg.InsecurePort != "" {
+		opts = append(opts, zitadel.WithInsecure(cfg.InsecurePort))
+	}
+	instance := zitadel.New(cfg.Domain, opts...)
+
+	discovery, err := client.Discover(ctx, instance.Origin(), httpClient)
+	if err != nil {
+		return nil, fmt.Errorf("zitadel: OIDC discovery at %s: %w", instance.Origin(), err)
+	}
+
+	keySet := rp.NewRemoteKeySet(httpClient, discovery.JwksURI)
+	verifier := op.NewAccessTokenVerifier(discovery.Issuer, keySet)
+
+	return &ZitadelVerifier{
+		instance:  instance,
+		verifier:  verifier,
+		clientIDs: append([]string(nil), cfg.ClientIDs...),
+	}, nil
+}
+
+// Origin returns the instance origin used for discovery, for logging.
+func (z *ZitadelVerifier) Origin() string {
+	if z == nil || z.instance == nil {
+		return ""
+	}
+	return z.instance.Origin()
+}
+
+// CheckAuthorization validates an "Authorization: Bearer <token>" header and
+// returns the SDK's verified introspection context.
+func (z *ZitadelVerifier) CheckAuthorization(ctx context.Context, authHeader string) (*oauth.IntrospectionContext, error) {
+	accessToken, ok := strings.CutPrefix(authHeader, oidc.BearerToken)
+	if !ok {
+		return nil, fmt.Errorf("%w: expected %q prefix", oauth.ErrInvalidAuthorizationHeader, oidc.BearerToken)
+	}
+	accessToken = strings.TrimSpace(accessToken)
+	if accessToken == "" {
+		return nil, fmt.Errorf("%w: empty bearer token", oauth.ErrInvalidAuthorizationHeader)
+	}
+
+	claims, err := op.VerifyAccessToken[*oidc.AccessTokenClaims](ctx, accessToken, z.verifier)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %v", oauth.ErrInvalidToken, err)
+	}
+
+	if len(claims.Audience) == 0 {
+		return nil, fmt.Errorf("%w: empty aud", oauth.ErrInvalidToken)
+	}
+	if !z.audienceAllowed(claims.Audience) {
+		return nil, fmt.Errorf("%w: %v", ErrAudienceNotAllowed, claims.Audience)
+	}
+
+	authCtx := &oauth.IntrospectionContext{
+		IntrospectionResponse: oidc.IntrospectionResponse{
+			Active:     true,
+			Issuer:     claims.Issuer,
+			Subject:    claims.Subject,
+			Audience:   claims.Audience,
+			Expiration: claims.Expiration,
+			IssuedAt:   claims.IssuedAt,
+			NotBefore:  claims.NotBefore,
+			ClientID:   claims.ClientID,
+			JWTID:      claims.JWTID,
+			Claims:     claims.Claims,
+		},
+	}
+	authCtx.SetToken(accessToken)
+	return authCtx, nil
+}
+
+// audienceAllowed reports whether any aud entry is in the allow-list.
+// An empty allow-list accepts every non-empty audience.
+func (z *ZitadelVerifier) audienceAllowed(audience []string) bool {
+	if len(z.clientIDs) == 0 {
+		return true
+	}
+	for _, aud := range audience {
+		if slices.Contains(z.clientIDs, aud) {
+			return true
+		}
+	}
+	return false
+}
@@ -0,0 +1,294 @@
+package auth
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"math/big"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/golang-jwt/jwt/v5"
+)
+
+// testIssuer mimics one ZITADEL instance: OIDC discovery on
+// /.well-known/openid-configuration and the JWKS on /oauth/v2/keys.
+type testIssuer struct {
+	server  *httptest.Server
+	key     *rsa.PrivateKey
+	kid     string
+	jwksHit int
+}
+
+func newTestIssuer(t *testing.T) *testIssuer {
+	t.Helper()
+
+	key, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+
+	ti := &testIssuer{key: key, kid: "test-kid-1"}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) {
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"issuer":   ti.server.URL,
+			"jwks_uri": ti.server.URL + "/oauth/v2/keys",
+		})
+	})
+	mux.HandleFunc("/oauth/v2/keys", func(w http.ResponseWriter, r *http.Request) {
+		ti.jwksHit++
+		pub := key.PublicKey
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"keys": []map[string]any{{
+				"kty": "RSA",
+				"kid": ti.kid,
+				"use": "sig",
+				"alg": "RS256",
+				"n":   base64.RawURLEncoding.EncodeToString(pub.N.Bytes()),
+				"e":   base64.RawURLEncoding.EncodeToString(big.NewInt(int64(pub.E)).Bytes()),
+			}},
+		})
+	})
+
+	ti.server = httptest.NewServer(mux)
+	t.Cleanup(ti.server.Close)
+	return ti
+}
+
+// sign mints an access token with the given audience and project roles.
+func (ti *testIssuer) sign(t *testing.T, aud []string, roles map[string]any) string {
+	t.Helper()
+
+	claims := jwt.MapClaims{
+		"iss": ti.server.URL,
+		"sub": "396757800000000001",
+		"exp": time.Now().Add(time.Hour).Unix(),
+		"iat": time.Now().Unix(),
+	}
+	if len(aud) > 0 {
+		audClaim := make([]interface{}, 0, len(aud))
+		for _, a := range aud {
+			audClaim = append(audClaim, a)
+		}
+		claims["aud"] = audClaim
+	}
+	if roles != nil {
+		claims["urn:zitadel:iam:org:project:roles"] = roles
+	}
+
+	tok := jwt.NewWithClaims(jwt.SigningMethodRS256, claims)
+	tok.Header["kid"] = ti.kid
+	signed, err := tok.SignedString(ti.key)
+	if err != nil {
+		t.Fatalf("sign token: %v", err)
+	}
+	return signed
+}
+
+// verifierFor builds a ZitadelVerifier pointed at the httptest instance.
+// The SDK is given the loopback host and its port, so discovery goes to the
+// test server rather than a real domain.
+func verifierFor(t *testing.T, ti *testIssuer, clientIDs []string) *ZitadelVerifier {
+	t.Helper()
+
+	host, port, ok := strings.Cut(strings.TrimPrefix(ti.server.URL, "http://"), ":")
+	if !ok {
+		t.Fatalf("unexpected test server URL %q", ti.server.URL)
+	}
+
+	v, err := NewZitadelVerifier(context.Background(), ZitadelConfig{
+		Domain:       host,
+		InsecurePort: port,
+		ClientIDs:    clientIDs,
+		HTTPClient:   ti.server.Client(),
+	})
+	if err != nil {
+		t.Fatalf("NewZitadelVerifier: %v", err)
+	}
+	return v
+}
+
+func TestZitadelVerifier_AcceptsTokenWithAllowedAudience(t *testing.T) {
+	ti := newTestIssuer(t)
+	v := verifierFor(t, ti, []string{"396757803559027071", "396757800874672511"})
+
+	token := ti.sign(t, []string{"396757803559027071"}, map[string]any{
+		"llm:inference": map[string]any{"396757796898471937": "homelab"},
+	})
+
+	authCtx, err := v.CheckAuthorization(context.Background(), "Bearer "+token)
+	if err != nil {
+		t.Fatalf("expected token to pass, got %v", err)
+	}
+	if !authCtx.IsAuthorized() {
+		t.Error("expected IsAuthorized() true")
+	}
+	if want := "396757800000000001"; authCtx.UserID() != want {
+		t.Errorf("UserID() = %q, want %q", authCtx.UserID(), want)
+	}
+	if !authCtx.IsGrantedRole("llm:inference") {
+		t.Error("expected SDK IsGrantedRole to see llm:inference from the project roles claim")
+	}
+	if ti.jwksHit == 0 {
+		t.Error("expected the SDK to fetch the JWKS")
+	}
+}
+
+func TestZitadelVerifier_RejectsForeignAudience(t *testing.T) {
+	ti := newTestIssuer(t)
+	v := verifierFor(t, ti, []string{"396757803559027071"})
+
+	token := ti.sign(t, []string{"999999999999999999"}, nil)
+
+	_, err := v.CheckAuthorization(context.Background(), "Bearer "+token)
+	if err == nil {
+		t.Fatal("expected rejection for an audience outside the allow-list")
+	}
+	if !strings.Contains(err.Error(), ErrAudienceNotAllowed.Error()) {
+		t.Errorf("expected ErrAudienceNotAllowed, got %v", err)
+	}
+}
+
+func TestZitadelVerifier_AcceptsAnyAudienceWhenAllowListEmpty(t *testing.T) {
+	ti := newTestIssuer(t)
+	v := verifierFor(t, ti, nil)
+
+	token := ti.sign(t, []string{"396757803877794175"}, nil)
+
+	if _, err := v.CheckAuthorization(context.Background(), "Bearer "+token); err != nil {
+		t.Fatalf("empty allow-list should accept any audience, got %v", err)
+	}
+}
+
+func TestZitadelVerifier_RejectsMissingBearerPrefix(t *testing.T) {
+	ti := newTestIssuer(t)
+	v := verifierFor(t, ti, nil)
+
+	token := ti.sign(t, []string{"396757803877794175"}, nil)
+
+	if _, err := v.CheckAuthorization(context.Background(), token); err == nil {
+		t.Fatal("expected rejection when the Bearer prefix is missing")
+	}
+}
+
+func TestZitadelVerifier_RejectsExpiredToken(t *testing.T) {
+	ti := newTestIssuer(t)
+	v := verifierFor(t, ti, nil)
+
+	claims := jwt.MapClaims{
+		"iss": ti.server.URL,
+		"sub": "u1",
+		"aud": []interface{}{"396757803877794175"},
+		"exp": time.Now().Add(-2 * time.Hour).Unix(),
+		"iat": time.Now().Add(-3 * time.Hour).Unix(),
+	}
+	tok := jwt.NewWithClaims(jwt.SigningMethodRS256, claims)
+	tok.Header["kid"] = ti.kid
+	signed, err := tok.SignedString(ti.key)
+	if err != nil {
+		t.Fatalf("sign token: %v", err)
+	}
+
+	if _, err := v.CheckAuthorization(context.Background(), "Bearer "+signed); err == nil {
+		t.Fatal("expected rejection for an expired token")
+	}
+}
+
+// TestValidator_SDKPathFeedsCheckPermissions proves the adapter from the SDK
+// introspection context to jwt.MapClaims keeps the capability gate working.
+func TestValidator_SDKPathFeedsCheckPermissions(t *testing.T) {
+	ti := newTestIssuer(t)
+	host, port, _ := strings.Cut(strings.TrimPrefix(ti.server.URL, "http://"), ":")
+
+	validator, err := NewZitadelValidator(context.Background(), ti.server.URL, []string{"396757803559027071"}, ZitadelConfig{
+		Domain:       host,
+		InsecurePort: port,
+		HTTPClient:   ti.server.Client(),
+	})
+	if err != nil {
+		t.Fatalf("NewZitadelValidator: %v", err)
+	}
+
+	good := ti.sign(t, []string{"396757803559027071"}, map[string]any{
+		"llm:inference": map[string]any{"396757796898471937": "homelab"},
+	})
+	claims, err := validator.ValidateBearerToken("Bearer " + good)
+	if err != nil {
+		t.Fatalf("ValidateBearerToken: %v", err)
+	}
+	if !validator.CheckPermissions(claims, "llm:inference") {
+		t.Error("expected CheckPermissions to accept llm:inference via the SDK path")
+	}
+	if validator.CheckPermissions(claims, "sqs:write") {
+		t.Error("expected CheckPermissions to reject a role the token does not carry")
+	}
+
+	noRole := ti.sign(t, []string{"396757803559027071"}, nil)
+	claims, err = validator.ValidateBearerToken("Bearer " + noRole)
+	if err != nil {
+		t.Fatalf("ValidateBearerToken without roles: %v", err)
+	}
+	if validator.CheckPermissions(claims, "llm:inference") {
+		t.Error("expected CheckPermissions to reject a token with no project roles")
+	}
+}
+
+func TestIssuerMatches(t *testing.T) {
+	cases := []struct {
+		iss        string
+		configured string
+		want       bool
+	}{
+		{"https://zitadel.riotpiao.com", "https://zitadel.riotpiao.com", true},
+		{"https://zitadel.riotpiao.com/", "https://zitadel.riotpiao.com", true},
+		{"https://zitadel.riotpiao.com/oauth/v2", "https://zitadel.riotpiao.com", true},
+		{"https://evil.example.com", "https://zitadel.riotpiao.com", false},
+		{"https://zitadel.riotpiao.com.evil.com", "https://zitadel.riotpiao.com", false},
+		{"anything", "", true},
+	}
+	for _, c := range cases {
+		if got := issuerMatches(c.iss, c.configured); got != c.want {
+			t.Errorf("issuerMatches(%q, %q) = %v, want %v", c.iss, c.configured, got, c.want)
+		}
+	}
+}
+
+func TestClaimsFromIntrospection_Nil(t *testing.T) {
+	if got := claimsFromIntrospection(nil); got == nil || len(got) != 0 {
+		t.Errorf("expected empty non-nil claims, got %v", got)
+	}
+}
+
+func TestAudienceValues(t *testing.T) {
+	cases := []struct {
+		name  string
+		claim any
+		want  int
+		ok    bool
+	}{
+		{"string", "api-gw", 1, true},
+		{"empty string", "", 1, false},
+		{"string slice", []string{"a", "b"}, 2, true},
+		{"iface slice", []interface{}{"a", 5}, 1, true},
+		{"missing", nil, 0, false},
+	}
+	for _, c := range cases {
+		claims := jwt.MapClaims{}
+		if c.claim != nil {
+			claims["aud"] = c.claim
+		}
+		got, ok := audienceValues(claims)
+		if ok != c.ok || len(got) != c.want {
+			t.Errorf("%s: audienceValues = (%v, %v), want len %d ok %v", c.name, got, ok, c.want, c.ok)
+		}
+	}
+	_ = fmt.Sprint()
+}
@@ -66,12 +66,23 @@ type AuthConfig struct {
 	JWKSURL string
 	// RequiredCapability is the permission required for LLM inference (e.g., "llm:inference").
 	RequiredCapability string
-	// TokenURL is the Authentik token endpoint for password/refresh grants.
+	// TokenURL is the OIDC token endpoint used for token exchange
+	// (client_credentials / password grants).
 	TokenURL string
 	// ClientID is the OAuth2 client ID for token exchange.
 	ClientID string
 	// ClientSecret is the OAuth2 client secret (loaded from env, never from config file).
 	ClientSecret string
+	// ZitadelDomain is the ZITADEL instance host (for example
+	// "zitadel.riotpiao.com"). When set, token validation uses the official
+	// ZITADEL Go SDK instead of the raw JWKS path.
+	ZitadelDomain string
+	// ZitadelInsecurePort, when non-empty, makes the SDK speak plain HTTP to
+	// that port. Only for a local port-forward.
+	ZitadelInsecurePort string
+	// ZitadelClientIDs is the audience allow-list for the SDK verifier. Empty
+	// accepts any audience issued by the same instance.
+	ZitadelClientIDs []string
 }

 // Route represents a single route and its upstream configuration.
@@ -20,13 +20,16 @@ type rawConfig struct {

 // rawAuth represents auth configuration in YAML.
 type rawAuth struct {
-	Enabled            bool   `yaml:"enabled"`
-	Issuer             string `yaml:"issuer"`
-	Audience           string `yaml:"audience"`
-	JWKSURL            string `yaml:"jwksUrl"`
-	RequiredCapability string `yaml:"requiredCapability"`
-	TokenURL           string `yaml:"tokenUrl"`
-	ClientID           string `yaml:"clientId"`
+	Enabled             bool     `yaml:"enabled"`
+	Issuer              string   `yaml:"issuer"`
+	Audience            string   `yaml:"audience"`
+	JWKSURL             string   `yaml:"jwksUrl"`
+	RequiredCapability  string   `yaml:"requiredCapability"`
+	TokenURL            string   `yaml:"tokenUrl"`
+	ClientID            string   `yaml:"clientId"`
+	ZitadelDomain       string   `yaml:"zitadelDomain"`
+	ZitadelInsecurePort string   `yaml:"zitadelInsecurePort"`
+	ZitadelClientIDs    []string `yaml:"zitadelClientIds"`
 }

 // rawRoute represents a single route in the YAML configuration.
@@ -184,14 +187,17 @@ func LoadRoutesAndModelsFromFile(path string) (map[string]*Route, map[string]*Mo

 	// Parse auth config
 	authConfig := AuthConfig{
-		Enabled:            raw.Auth.Enabled,
-		Issuer:             raw.Auth.Issuer,
-		Audience:           raw.Auth.Audience,
-		JWKSURL:            raw.Auth.JWKSURL,
-		RequiredCapability: raw.Auth.RequiredCapability,
-		TokenURL:           raw.Auth.TokenURL,
-		ClientID:           raw.Auth.ClientID,
-		ClientSecret:       os.Getenv("AUTH_CLIENT_SECRET"),
+		Enabled:             raw.Auth.Enabled,
+		Issuer:              raw.Auth.Issuer,
+		Audience:            raw.Auth.Audience,
+		JWKSURL:             raw.Auth.JWKSURL,
+		RequiredCapability:  raw.Auth.RequiredCapability,
+		TokenURL:            raw.Auth.TokenURL,
+		ClientID:            raw.Auth.ClientID,
+		ClientSecret:        os.Getenv("AUTH_CLIENT_SECRET"),
+		ZitadelDomain:       raw.Auth.ZitadelDomain,
+		ZitadelInsecurePort: raw.Auth.ZitadelInsecurePort,
+		ZitadelClientIDs:    raw.Auth.ZitadelClientIDs,
 	}

 	return routes, models, adapters, authConfig, nil
@@ -15,7 +15,7 @@ type Metrics struct {

 	// Request latencies: request_duration_seconds (histogram)
 	// Stored as cumulative buckets for Prometheus text format
-	requestDuration map[string]int64 // stores duration samples in milliseconds
+	requestDuration        map[string]int64             // stores duration samples in milliseconds
 	requestDurationBuckets map[string]map[float64]int64 // histogram buckets

 	// Bytes counters: gateway_bytes{direction, route, upstream}
@@ -41,22 +41,33 @@ type Metrics struct {
 	warmupRequestsDelayed int64
 	prefixCacheHitTokens  int64
 	queueFullRequests     int64
+
+	// Admission control metrics (#64)
+	admissionParked   int64
+	admissionRejected map[string]int64
+	engineStateStale  int64
+
+	// Engine gauge readings mirrored from vLLM (#64)
+	vllmRequestsRunning int64
+	vllmKVUsagePerc     float64
+	observerDegraded    int
 }

 // NewMetrics creates a new Metrics instance.
 func NewMetrics() *Metrics {
 	return &Metrics{
-		requestTotal:           make(map[string]int64),
-		requestDuration:        make(map[string]int64),
-		requestDurationBuckets: make(map[string]map[float64]int64),
-		bytesIn:                make(map[string]int64),
-		bytesOut:               make(map[string]int64),
-		upstreamHealth:         make(map[string]int),
+		requestTotal:            make(map[string]int64),
+		requestDuration:         make(map[string]int64),
+		requestDurationBuckets:  make(map[string]map[float64]int64),
+		bytesIn:                 make(map[string]int64),
+		bytesOut:                make(map[string]int64),
+		upstreamHealth:          make(map[string]int),
 		streamingResponsesTotal: make(map[string]int64),
-		streamingByteCount:     make(map[string]int64),
-		ttftMs:                 make(map[string][]int64),
-		itlMs:                  make(map[string][]int64),
-		tokenCount:             make(map[string]int64),
+		streamingByteCount:      make(map[string]int64),
+		ttftMs:                  make(map[string][]int64),
+		itlMs:                   make(map[string][]int64),
+		tokenCount:              make(map[string]int64),
+		admissionRejected:       make(map[string]int64),
 	}
 }

@@ -288,6 +299,12 @@ func (m *Metrics) Reset() {
 	m.warmupRequestsDelayed = 0
 	m.prefixCacheHitTokens = 0
 	m.queueFullRequests = 0
+	m.admissionParked = 0
+	m.admissionRejected = make(map[string]int64)
+	m.engineStateStale = 0
+	m.vllmRequestsRunning = 0
+	m.vllmKVUsagePerc = 0
+	m.observerDegraded = 0
 }

 // IncrQueueFull increments queue_full_requests_total.
@@ -332,6 +349,98 @@ func (m *Metrics) GetPrefixCacheHitTokens() int64 {
 	return m.prefixCacheHitTokens
 }

+// IncrAdmissionParked increments admission_parked_total.
+func (m *Metrics) IncrAdmissionParked() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.admissionParked++
+}
+
+// GetAdmissionParked returns admission_parked_total.
+func (m *Metrics) GetAdmissionParked() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.admissionParked
+}
+
+// IncrAdmissionRejected increments admission_rejected_total{reason=reason}.
+func (m *Metrics) IncrAdmissionRejected(reason string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.admissionRejected[reason]++
+}
+
+// GetAdmissionRejected returns a copy of admission_rejected_total by reason.
+func (m *Metrics) GetAdmissionRejected() map[string]int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	out := make(map[string]int64, len(m.admissionRejected))
+	for reason, count := range m.admissionRejected {
+		out[reason] = count
+	}
+	return out
+}
+
+// IncrEngineStateStale increments engine_state_stale_total.
+func (m *Metrics) IncrEngineStateStale() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.engineStateStale++
+}
+
+// GetEngineStateStale returns engine_state_stale_total.
+func (m *Metrics) GetEngineStateStale() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.engineStateStale
+}
+
+// SetVLLMRequestsRunning sets vllm_requests_running from the gateway's own reading.
+func (m *Metrics) SetVLLMRequestsRunning(n int64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.vllmRequestsRunning = n
+}
+
+// GetVLLMRequestsRunning returns vllm_requests_running.
+func (m *Metrics) GetVLLMRequestsRunning() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.vllmRequestsRunning
+}
+
+// SetVLLMKVUsagePerc sets vllm_kv_usage_perc from the gateway's own reading.
+func (m *Metrics) SetVLLMKVUsagePerc(v float64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.vllmKVUsagePerc = v
+}
+
+// GetVLLMKVUsagePerc returns vllm_kv_usage_perc.
+func (m *Metrics) GetVLLMKVUsagePerc() float64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.vllmKVUsagePerc
+}
+
+// SetObserverDegraded sets observer_degraded (1 = startup self-check failed).
+func (m *Metrics) SetObserverDegraded(degraded bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if degraded {
+		m.observerDegraded = 1
+	} else {
+		m.observerDegraded = 0
+	}
+}
+
+// GetObserverDegraded returns observer_degraded.
+func (m *Metrics) GetObserverDegraded() int {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.observerDegraded
+}
+
 // globalMetrics is the singleton Metrics instance for gateway-wide telemetry.
 var (
 	globalMetrics     *Metrics
@@ -48,6 +48,38 @@ func (p *PrometheusExporter) Export() string {
 	lines = append(lines, "# TYPE vllm_prefix_cache_hit_tokens_total counter")
 	lines = append(lines, fmt.Sprintf("vllm_prefix_cache_hit_tokens_total %d", p.metrics.GetPrefixCacheHitTokens()))

+	lines = append(lines, "# HELP admission_parked_total Requests parked waiting for engine admission")
+	lines = append(lines, "# TYPE admission_parked_total counter")
+	lines = append(lines, fmt.Sprintf("admission_parked_total %d", p.metrics.GetAdmissionParked()))
+
+	lines = append(lines, "# HELP admission_rejected_total Requests rejected by engine admission control")
+	lines = append(lines, "# TYPE admission_rejected_total counter")
+	rejected := p.metrics.GetAdmissionRejected()
+	reasons := make([]string, 0, len(rejected))
+	for reason := range rejected {
+		reasons = append(reasons, reason)
+	}
+	sort.Strings(reasons)
+	for _, reason := range reasons {
+		lines = append(lines, fmt.Sprintf(`admission_rejected_total{reason="%s"} %d`, reason, rejected[reason]))
+	}
+
+	lines = append(lines, "# HELP engine_state_stale_total Times the engine metrics feed went stale")
+	lines = append(lines, "# TYPE engine_state_stale_total counter")
+	lines = append(lines, fmt.Sprintf("engine_state_stale_total %d", p.metrics.GetEngineStateStale()))
+
+	lines = append(lines, "# HELP vllm_requests_running vLLM requests running as read by the gateway")
+	lines = append(lines, "# TYPE vllm_requests_running gauge")
+	lines = append(lines, fmt.Sprintf("vllm_requests_running %d", p.metrics.GetVLLMRequestsRunning()))
+
+	lines = append(lines, "# HELP vllm_kv_usage_perc vLLM KV cache usage fraction as read by the gateway")
+	lines = append(lines, "# TYPE vllm_kv_usage_perc gauge")
+	lines = append(lines, fmt.Sprintf("vllm_kv_usage_perc %g", p.metrics.GetVLLMKVUsagePerc()))
+
+	lines = append(lines, "# HELP observer_degraded Engine metrics observer startup self-check failed (1) or healthy (0)")
+	lines = append(lines, "# TYPE observer_degraded gauge")
+	lines = append(lines, fmt.Sprintf("observer_degraded %d", p.metrics.GetObserverDegraded()))
+
 	return strings.Join(lines, "\n") + "\n"
 }

@@ -0,0 +1,333 @@
+package proxy
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/logging"
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
+)
+
+// Metric names published by the vLLM engine. Admission is gated on these
+// rather than on a per-pod semaphore: every gateway replica reads the same
+// engine, so the limit is global by construction (#64).
+const (
+	metricRequestsRunning = "vllm:num_requests_running"
+	metricRequestsWaiting = "vllm:num_requests_waiting"
+	metricKVCacheUsage    = "vllm:kv_cache_usage_perc"
+
+	// enginePollInterval is how often each backend's /metrics is scraped.
+	enginePollInterval = 2 * time.Second
+
+	// engineStaleThreshold is the number of consecutive failed polls after
+	// which the state is marked unhealthy and admission fails open.
+	engineStaleThreshold = 3
+
+	// engineMetricsTimeout bounds a single scrape.
+	engineMetricsTimeout = 2 * time.Second
+)
+
+// EngineState is the gateway's view of one vLLM engine, mirrored from its
+// own metrics endpoint.
+type EngineState struct {
+	Running   int64
+	Waiting   int64
+	KVUsage   float64
+	UpdatedAt time.Time
+	Healthy   bool
+}
+
+// InFlight returns the number of requests the engine currently holds:
+// running plus waiting.
+func (s EngineState) InFlight() int64 {
+	return s.Running + s.Waiting
+}
+
+// EngineObserver polls one vLLM backend's /metrics and publishes the gauges
+// admission is gated on. Healthy is false until the first successful poll and
+// after engineStaleThreshold consecutive failures.
+type EngineObserver struct {
+	addr     string
+	interval time.Duration
+	client   *http.Client
+
+	mu       sync.Mutex
+	state    EngineState
+	failures int
+	degraded bool
+	tick     chan struct{}
+}
+
+// NewEngineObserver creates an observer for the backend at addr.
+func NewEngineObserver(addr string, interval time.Duration) *EngineObserver {
+	return &EngineObserver{
+		addr:     addr,
+		interval: interval,
+		// Own client: the proxy transports carry socket options and connect
+		// timeouts tuned for inference, not for a short metrics scrape.
+		client: &http.Client{Timeout: engineMetricsTimeout},
+		tick:   make(chan struct{}),
+	}
+}
+
+// Start launches the polling goroutine. It returns immediately; the goroutine
+// exits when ctx is cancelled.
+func (o *EngineObserver) Start(ctx context.Context) {
+	go func() {
+		ticker := time.NewTicker(o.interval)
+		defer ticker.Stop()
+		// The first scrape waits one interval rather than firing at Start().
+		// Scraping immediately makes the gateway contact every observed backend
+		// before any request needs one, which is wasted work and a surprising
+		// side effect for anything asserting on upstream traffic. State is
+		// unknown until the first poll, and unknown fails open, so the deferral
+		// costs at most one interval of ungated admission. The startup
+		// self-check (decision Q21) still runs, on that first tick.
+		startup := true
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				o.poll(ctx, startup)
+				startup = false
+			}
+		}
+	}()
+}
+
+// Get returns the most recent engine state.
+func (o *EngineObserver) Get() EngineState {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.state
+}
+
+// Tick returns a channel that is closed and replaced after every successful
+// poll, waking requests parked on engine admission.
+func (o *EngineObserver) Tick() <-chan struct{} {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.tick
+}
+
+// Degraded reports whether the startup self-check found the metrics feed
+// unusable (parse failure or missing series).
+func (o *EngineObserver) Degraded() bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.degraded
+}
+
+// Snapshot returns the latest state together with the tick channel that the
+// next successful poll will close. Both are read under one lock on purpose:
+// calling Get and Tick separately can miss the wakeup that lands between the
+// two calls, parking a request for an extra poll interval.
+func (o *EngineObserver) Snapshot() (EngineState, <-chan struct{}) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.state, o.tick
+}
+
+// markDegraded records whether the metrics feed is currently unusable and
+// mirrors that into observer_degraded. It is cleared on the next parseable
+// scrape, so one truncated response cannot pin the gauge for the process
+// lifetime while metric-name drift after the first poll is still reported.
+func (o *EngineObserver) markDegraded(v bool) {
+	o.mu.Lock()
+	changed := o.degraded != v
+	o.degraded = v
+	o.mu.Unlock()
+	if changed {
+		observability.Global().SetObserverDegraded(v)
+	}
+}
+
+// poll scrapes /metrics once. startup marks the first poll, whose failure is a
+// self-diagnosis signal rather than a transient blip.
+func (o *EngineObserver) poll(ctx context.Context, startup bool) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+o.addr+"/metrics", nil)
+	if err != nil {
+		o.recordFailure()
+		logging.Warnf("engine metrics request build failed", map[string]string{
+			"address": o.addr,
+			"err":     err.Error(),
+		})
+		return
+	}
+	resp, err := o.client.Do(req)
+	if err != nil {
+		o.recordFailure()
+		logging.Warnf("engine metrics scrape failed", map[string]string{
+			"address": o.addr,
+			"err":     err.Error(),
+		})
+		return
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		o.recordFailure()
+		logging.Warnf("engine metrics scrape returned non-200", map[string]string{
+			"address": o.addr,
+			"status":  strconv.Itoa(resp.StatusCode),
+		})
+		return
+	}
+
+	running, waiting, kvUsage, err := ParseEngineMetrics(resp.Body)
+	if err != nil {
+		// Metric-name drift must be loud, never a silent fall-back to fail-open
+		// admission (decision Q21). The gauge tracks current state, so it clears
+		// on the next parseable scrape instead of pinning for the pod's life.
+		o.markDegraded(true)
+		if startup {
+			logging.Errorf("engine metrics unusable at startup", err, map[string]string{
+				"address": o.addr,
+			})
+		} else {
+			logging.Warnf("engine metrics parse failed", map[string]string{
+				"address": o.addr,
+				"err":     err.Error(),
+			})
+		}
+		o.recordFailure()
+		return
+	}
+
+	o.markDegraded(false)
+	o.recordSuccess(running, waiting, kvUsage)
+}
+
+// recordSuccess stores a fresh reading and wakes parked waiters.
+func (o *EngineObserver) recordSuccess(running, waiting int64, kvUsage float64) {
+	o.mu.Lock()
+	o.state = EngineState{
+		Running:   running,
+		Waiting:   waiting,
+		KVUsage:   kvUsage,
+		UpdatedAt: time.Now(),
+		Healthy:   true,
+	}
+	o.failures = 0
+	close(o.tick)
+	o.tick = make(chan struct{})
+	o.mu.Unlock()
+
+	m := observability.Global()
+	m.SetVLLMRequestsRunning(running)
+	m.SetVLLMKVUsagePerc(kvUsage)
+}
+
+// recordFailure counts a failed poll and, on the third consecutive failure,
+// marks the state unhealthy and counts the staleness transition once — not
+// once per failed poll. Last known values are kept.
+func (o *EngineObserver) recordFailure() {
+	o.mu.Lock()
+	o.failures++
+	stale := o.failures == engineStaleThreshold
+	if o.failures >= engineStaleThreshold {
+		o.state.Healthy = false
+	}
+	failures := o.failures
+	o.mu.Unlock()
+
+	if stale {
+		observability.Global().IncrEngineStateStale()
+		logging.Warnf("engine metrics stale; admission is failing open", map[string]string{
+			"address":  o.addr,
+			"failures": strconv.Itoa(failures),
+		})
+	}
+}
+
+// ParseEngineMetrics extracts the running, waiting and KV-cache-usage gauges
+// from a vLLM Prometheus exposition body. Comment lines, unrelated series and
+// non-finite samples are ignored; values are summed across repeated label sets.
+// It returns an error naming any of the three series that is absent.
+func ParseEngineMetrics(r io.Reader) (running int64, waiting int64, kvUsage float64, err error) {
+	scanner := bufio.NewScanner(r)
+	// The real exposition body is large; give the scanner room per line.
+	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+
+	var hasRunning, hasWaiting, hasKV bool
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+		name, value, ok := splitMetricSample(line)
+		if !ok {
+			continue
+		}
+		switch name {
+		case metricRequestsRunning, metricRequestsWaiting, metricKVCacheUsage:
+		default:
+			continue
+		}
+		f, err := strconv.ParseFloat(value, 64)
+		if err != nil || math.IsNaN(f) || math.IsInf(f, 0) {
+			continue
+		}
+		switch name {
+		case metricRequestsRunning:
+			running += int64(f)
+			hasRunning = true
+		case metricRequestsWaiting:
+			waiting += int64(f)
+			hasWaiting = true
+		case metricKVCacheUsage:
+			kvUsage += f
+			hasKV = true
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return 0, 0, 0, fmt.Errorf("read engine metrics: %w", err)
+	}
+
+	var missing []string
+	if !hasRunning {
+		missing = append(missing, metricRequestsRunning)
+	}
+	if !hasWaiting {
+		missing = append(missing, metricRequestsWaiting)
+	}
+	if !hasKV {
+		missing = append(missing, metricKVCacheUsage)
+	}
+	if len(missing) > 0 {
+		return 0, 0, 0, fmt.Errorf("engine metrics missing series: %s", strings.Join(missing, ", "))
+	}
+	return running, waiting, kvUsage, nil
+}
+
+// splitMetricSample splits one Prometheus exposition line into its metric name
+// and its first sample value. Lines without a value are reported as not ok.
+func splitMetricSample(line string) (name, value string, ok bool) {
+	end := strings.IndexAny(line, "{ \t")
+	if end <= 0 {
+		return "", "", false
+	}
+	name = line[:end]
+	rest := line[end:]
+	if rest[0] == '{' {
+		closeBrace := strings.LastIndex(rest, "}")
+		if closeBrace < 0 {
+			return "", "", false
+		}
+		rest = rest[closeBrace+1:]
+	}
+	fields := strings.Fields(rest)
+	if len(fields) == 0 {
+		return "", "", false
+	}
+	return name, fields[0], true
+}
@@ -3,7 +3,6 @@ package proxy
 import (
 	"bufio"
 	"fmt"
-	"io"
 	"net"
 	"net/http"
 	"strings"
@@ -12,87 +11,37 @@ import (
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
 )

-// LLMMetricsCapture wraps a response writer to capture TTFT and ITL metrics
-type LLMMetricsCapture struct {
-	writer              io.WriteCloser
-	model               string
-	metrics             *observability.Metrics
-	firstTokenTime      time.Time
-	lastTokenTime       time.Time
-	requestStartTime    time.Time
-	ttftRecorded        bool
-	tokenCount          int64
-	responseStartTime   time.Time
-}
-
-// NewLLMMetricsCapture creates a new metrics capture wrapper
-func NewLLMMetricsCapture(writer io.WriteCloser, model string, metrics *observability.Metrics, startTime time.Time) *LLMMetricsCapture {
-	return &LLMMetricsCapture{
-		writer:            writer,
-		model:             model,
-		metrics:           metrics,
-		requestStartTime:  startTime,
-		responseStartTime: time.Now(),
-	}
-}
-
-// Write intercepts writes to detect tokens and record metrics
-func (c *LLMMetricsCapture) Write(p []byte) (int, error) {
-	// Record first token time
-	if !c.ttftRecorded && len(p) > 0 {
-		now := time.Now()
-		ttft := now.Sub(c.requestStartTime).Milliseconds()
-		c.metrics.RecordTTFT(c.model, ttft)
-		c.ttftRecorded = true
-		c.firstTokenTime = now
-		c.lastTokenTime = now
-	}
-
-	// Count tokens in SSE stream (simple: count "data: " lines)
-	if c.ttftRecorded {
-		tokenCount := strings.Count(string(p), "data: ")
-		if tokenCount > 0 {
-			now := time.Now()
-			if !c.firstTokenTime.IsZero() && c.lastTokenTime != now {
-				itl := now.Sub(c.lastTokenTime).Milliseconds()
-				c.metrics.RecordITL(c.model, itl)
-			}
-			c.lastTokenTime = now
-			c.tokenCount += int64(tokenCount)
-		}
-	}
-
-	return c.writer.Write(p)
-}
-
-// Close records final metrics and closes writer
-func (c *LLMMetricsCapture) Close() error {
-	if c.tokenCount > 0 {
-		c.metrics.RecordTokenCount(c.model, c.tokenCount)
-	}
-	return c.writer.Close()
-}
-
-// ResponseWriterWrapper wraps http.ResponseWriter to capture metrics
+// ResponseWriterWrapper wraps an http.ResponseWriter to capture per-request LLM
+// latency metrics (TTFT, ITL, generated tokens) while a streaming response is
+// copied from the upstream by httputil.ReverseProxy.
+//
+// Only server-sent-event responses are measured. A non-streaming completion is
+// written in a single Write, so recording it would emit one "TTFT" sample equal
+// to the whole request latency and no ITL samples at all — numbers that look
+// plausible but mean nothing.
 type ResponseWriterWrapper struct {
-	writer          http.ResponseWriter
-	statusCode      int
-	metrics         *observability.Metrics
-	model           string
-	startTime       time.Time
-	firstByteTime   time.Time
-	lastWriteTime   time.Time
-	ttftRecorded    bool
+	writer        http.ResponseWriter
+	statusCode    int
+	metrics       *observability.Metrics
+	model         string
+	startTime     time.Time
+	firstByteTime time.Time
+	lastWriteTime time.Time
+	ttftRecorded  bool
+	streamChecked bool
+	streaming     bool
 }

-// NewResponseWriterWrapper creates a wrapper for response writer
+// NewResponseWriterWrapper creates a wrapper for response writer.
+// startTime must be the time the request entered the gateway, not the time this
+// wrapper was constructed, otherwise TTFT measures the wrong interval.
 func NewResponseWriterWrapper(w http.ResponseWriter, model string, metrics *observability.Metrics, startTime time.Time) *ResponseWriterWrapper {
 	return &ResponseWriterWrapper{
-		writer:      w,
-		model:       model,
-		metrics:     metrics,
-		startTime:   startTime,
-		statusCode:  200,
+		writer:     w,
+		model:      model,
+		metrics:    metrics,
+		startTime:  startTime,
+		statusCode: 200,
 	}
 }

@@ -101,29 +50,48 @@ func (w *ResponseWriterWrapper) Header() http.Header {
 	return w.writer.Header()
 }

+// isStreaming reports whether this is an SSE response. Resolved once from the
+// Content-Type header, which ReverseProxy has already set before the first write.
+func (w *ResponseWriterWrapper) isStreaming() bool {
+	if !w.streamChecked {
+		w.streamChecked = true
+		w.streaming = strings.Contains(w.writer.Header().Get("Content-Type"), "text/event-stream")
+	}
+	return w.streaming
+}
+
+// countSSETokens counts generated tokens in a chunk of an SSE stream. These
+// upstreams emit one `data:` event per token, so the event count is the token
+// count; the `[DONE]` terminator is not a token.
+func countSSETokens(b []byte) int64 {
+	var n int64
+	for _, line := range strings.Split(string(b), "\n") {
+		line = strings.TrimSpace(line)
+		if strings.HasPrefix(line, "data:") && !strings.HasPrefix(line, "data: [DONE]") {
+			n++
+		}
+	}
+	return n
+}
+
 // Write implements http.ResponseWriter
 func (w *ResponseWriterWrapper) Write(b []byte) (int, error) {
-	// Record TTFT on first write
-	if !w.ttftRecorded && len(b) > 0 {
+	if len(b) > 0 && w.isStreaming() {
 		now := time.Now()
-		ttft := now.Sub(w.startTime).Milliseconds()
-		w.metrics.RecordTTFT(w.model, ttft)
-		w.ttftRecorded = true
-		w.firstByteTime = now
-		w.lastWriteTime = now
-	}
-
-	// Record ITL for subsequent writes (for streaming)
-	if w.ttftRecorded && len(b) > 0 {
-		now := time.Now()
-		if !w.firstByteTime.IsZero() && w.lastWriteTime != now {
-			itl := now.Sub(w.lastWriteTime).Milliseconds()
-			// Only record if ITL > 0 (avoid recording same millisecond twice)
-			if itl > 0 {
-				w.metrics.RecordITL(w.model, itl)
-			}
+		if !w.ttftRecorded {
+			w.metrics.RecordTTFT(w.model, now.Sub(w.startTime).Milliseconds())
+			w.ttftRecorded = true
+			w.firstByteTime = now
+		} else if itl := now.Sub(w.lastWriteTime).Milliseconds(); itl > 0 {
+			// Only record positive ITL: a zero means two chunks landed in the
+			// same millisecond, which carries no timing information.
+			w.metrics.RecordITL(w.model, itl)
 		}
 		w.lastWriteTime = now
+
+		if n := countSSETokens(b); n > 0 {
+			w.metrics.RecordTokenCount(w.model, n)
+		}
 	}

 	return w.writer.Write(b)
@@ -150,4 +118,8 @@ func (w *ResponseWriterWrapper) Hijack() (net.Conn, *bufio.ReadWriter, error) {
 	return nil, nil, fmt.Errorf("response writer does not implement Hijacker")
 }

-
+// Unwrap exposes the underlying writer to http.ResponseController so optional
+// capabilities (Flush, Hijack, deadlines) keep working through the wrapper.
+func (w *ResponseWriterWrapper) Unwrap() http.ResponseWriter {
+	return w.writer
+}
@@ -0,0 +1,171 @@
+package proxy
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
+)
+
+// exportMetrics renders the Prometheus text exposition for m.
+func exportMetrics(m *observability.Metrics) string {
+	return observability.NewPrometheusExporter(m).Export()
+}
+
+// requireSeries asserts the exporter emitted a sample line with the exact value.
+func requireSeries(t *testing.T, out, want string) {
+	t.Helper()
+	for _, line := range strings.Split(out, "\n") {
+		if strings.TrimSpace(line) == want {
+			return
+		}
+	}
+	t.Errorf("missing metrics series %q\n--- got ---\n%s", want, out)
+}
+
+// TestResponseWriterWrapperRecordsSSEMetrics guards the regression where the
+// LLM latency wrapper existed but was never constructed, leaving
+// llm_ttft_seconds / llm_itl_seconds / llm_tokens_total as HELP-only stubs.
+func TestResponseWriterWrapperRecordsSSEMetrics(t *testing.T) {
+	m := observability.NewMetrics()
+
+	rec := httptest.NewRecorder()
+	rec.Header().Set("Content-Type", "text/event-stream")
+
+	w := NewResponseWriterWrapper(rec, "reasoning", m, time.Now().Add(-50*time.Millisecond))
+
+	if _, err := w.Write([]byte("data: {\"token\":\"a\"}\n\n")); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	time.Sleep(5 * time.Millisecond)
+	if _, err := w.Write([]byte("data: {\"token\":\"b\"}\n\n")); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	// The terminator is not a generated token.
+	if _, err := w.Write([]byte("data: [DONE]\n\n")); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	out := exportMetrics(m)
+	requireSeries(t, out, `llm_ttft_seconds_count{model="reasoning"} 1`)
+	requireSeries(t, out, `llm_itl_seconds_count{model="reasoning"} 1`)
+	requireSeries(t, out, `llm_tokens_total{model="reasoning"} 2`)
+}
+
+// TestResponseWriterWrapperIgnoresNonStreaming verifies a single-shot JSON
+// response does not produce a bogus TTFT sample equal to full request latency.
+func TestResponseWriterWrapperIgnoresNonStreaming(t *testing.T) {
+	m := observability.NewMetrics()
+
+	rec := httptest.NewRecorder()
+	rec.Header().Set("Content-Type", "application/json")
+
+	w := NewResponseWriterWrapper(rec, "reasoning", m, time.Now().Add(-50*time.Millisecond))
+	if _, err := w.Write([]byte(`{"choices":[]}`)); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	out := exportMetrics(m)
+	if strings.Contains(out, `llm_ttft_seconds_count{model="reasoning"}`) {
+		t.Errorf("non-streaming response recorded TTFT\n--- got ---\n%s", out)
+	}
+	if strings.Contains(out, `llm_tokens_total{model="reasoning"}`) {
+		t.Errorf("non-streaming response recorded tokens\n--- got ---\n%s", out)
+	}
+}
+
+// TestResponseWriterWrapperPreservesStreaming verifies the wrapper stays
+// transparent: bytes reach the client and Flush still works, so wrapping the
+// writer cannot silently break SSE (issue #33).
+func TestResponseWriterWrapperPreservesStreaming(t *testing.T) {
+	m := observability.NewMetrics()
+
+	rec := httptest.NewRecorder()
+	rec.Header().Set("Content-Type", "text/event-stream")
+
+	w := NewResponseWriterWrapper(rec, "reasoning", m, time.Now())
+	payload := "data: {\"token\":\"a\"}\n\n"
+	if _, err := w.Write([]byte(payload)); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	w.Flush()
+
+	if got := rec.Body.String(); got != payload {
+		t.Errorf("body = %q, want %q", got, payload)
+	}
+	if !rec.Flushed {
+		t.Error("Flush did not reach the underlying writer")
+	}
+}
+
+// TestHandlerRecordsLLMMetricsForStreaming is the end-to-end guard: it drives a
+// real streaming request through the Handler, so it fails if the wrapper is ever
+// constructed nowhere (the original defect) rather than merely being broken.
+func TestHandlerRecordsLLMMetricsForStreaming(t *testing.T) {
+	// Unique model name keeps the observations isolated in the global registry.
+	model := fmt.Sprintf("metrics-test-%d", time.Now().UnixNano())
+
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		rc := http.NewResponseController(w)
+		for _, tok := range []string{"a", "b", "c"} {
+			time.Sleep(2 * time.Millisecond) // guarantee distinct ITL samples
+			fmt.Fprintf(w, "data: {\"token\":%q}\n\n", tok)
+			if err := rc.Flush(); err != nil {
+				return
+			}
+		}
+		fmt.Fprint(w, "data: [DONE]\n\n")
+		_ = rc.Flush()
+	}))
+	defer upstream.Close()
+
+	cfg := &config.Config{
+		Models: map[string]*config.ModelUpstream{
+			model: {
+				Name:    model,
+				Address: strings.TrimPrefix(upstream.URL, "http://"),
+				Path:    "/v1/chat/completions",
+			},
+		},
+		Routes: make(map[string]*config.Route),
+	}
+
+	handler := New(cfg)
+	defer handler.Close()
+	srv := httptest.NewServer(handler)
+	defer srv.Close()
+
+	body := fmt.Sprintf(`{"model":%q,"stream":true,"messages":[{"role":"user","content":"hi"}]}`, model)
+	resp, err := http.Post(srv.URL+"/v1/chat/completions", "application/json", strings.NewReader(body))
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200", resp.StatusCode)
+	}
+	if _, err := io.ReadAll(resp.Body); err != nil {
+		t.Fatalf("read body: %v", err)
+	}
+
+	out := exportMetrics(observability.Global())
+	requireSeries(t, out, fmt.Sprintf(`llm_ttft_seconds_count{model=%q} 1`, model))
+	requireSeries(t, out, fmt.Sprintf(`llm_tokens_total{model=%q} 3`, model))
+
+	itlCount := regexp.MustCompile(fmt.Sprintf(`llm_itl_seconds_count\{model=%q\} (\d+)`, model))
+	match := itlCount.FindStringSubmatch(out)
+	if match == nil {
+		t.Errorf("no llm_itl_seconds samples recorded for %s\n--- got ---\n%s", model, out)
+	} else if match[1] == "0" {
+		t.Errorf("llm_itl_seconds_count = 0, want > 0")
+	}
+}
@@ -2,6 +2,7 @@
 package proxy

 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -47,6 +48,10 @@ type Handler struct {
 	schedulers  map[string]*Scheduler
 	// CAG: delay applied to warmup requests before queue entry (#45)
 	warmupDelay time.Duration
+	// engineObservers polls each model backend's /metrics for admission (#64)
+	engineObservers map[string]*EngineObserver
+	// maxBodyBytes caps synthesized model-route request bodies (MAX_BODY_BYTES)
+	maxBodyBytes int64
 	// transportsMu guards h.transports for goroutine-safe reads/writes.
 	transportsMu sync.RWMutex
 }
@@ -93,8 +98,29 @@ func New(cfg *config.Config) *Handler {
 		defaultMaxBodySize:    100 * 1024 * 1024,
 	}

-	// Initialize JWT validator if auth is enabled
-	if cfg.Auth.Enabled && cfg.Auth.JWKSURL != "" {
+	// Initialize JWT validator if auth is enabled.
+	//
+	// Two paths, in order of preference:
+	//  1. ZitadelDomain set -> official ZITADEL Go SDK (discovery + remote JWKS
+	//     + op.AccessTokenVerifier). This is the production path.
+	//  2. JWKSURL set -> legacy raw-JWKS validation, kept for tests and for
+	//     any deployment that has not moved to ZITADEL yet.
+	if cfg.Auth.Enabled && cfg.Auth.ZitadelDomain != "" {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		v, err := auth.NewZitadelValidator(ctx, cfg.Auth.Issuer, cfg.Auth.ZitadelClientIDs, auth.ZitadelConfig{
+			Domain:       cfg.Auth.ZitadelDomain,
+			InsecurePort: cfg.Auth.ZitadelInsecurePort,
+			ClientIDs:    cfg.Auth.ZitadelClientIDs,
+		})
+		if err != nil {
+			logging.Errorf("zitadel: verifier init failed; auth disabled until restart", err, map[string]string{
+				"domain": cfg.Auth.ZitadelDomain,
+			})
+		} else {
+			h.jwtValidator = v
+		}
+	} else if cfg.Auth.Enabled && cfg.Auth.JWKSURL != "" {
 		h.jwtValidator = auth.NewValidator(
 			cfg.Auth.Issuer,
 			cfg.Auth.Audience,
@@ -122,18 +148,80 @@ func New(cfg *config.Config) *Handler {
 			maxSlots = n
 		}
 	}
-	maxQueue := 20
+	maxQueue := 8
 	if v := os.Getenv("VLLM_MAX_QUEUE_DEPTH"); v != "" {
 		if n, err := strconv.Atoi(v); err == nil && n > 0 {
 			maxQueue = n
 		}
 	}
+
+	// Issue #64: engine-wide admission thresholds. MaxInFlight defaults to the
+	// same value as VLLM_MAX_NUM_SEQS; the observers below read the engine's
+	// own gauges, which every replica shares.
+	admissionMaxInFlight := maxSlots
+	if v := os.Getenv("ADMISSION_MAX_INFLIGHT"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			admissionMaxInFlight = n
+		}
+	}
+	admissionMaxKVUsage := 0.90
+	if v := os.Getenv("ADMISSION_MAX_KV_USAGE"); v != "" {
+		if f, err := strconv.ParseFloat(v, 64); err == nil && f > 0 {
+			admissionMaxKVUsage = f
+		}
+	}
+	maxParkSeconds := 120
+	if v := os.Getenv("ADMISSION_MAX_PARK_SECONDS"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			maxParkSeconds = n
+		}
+	}
+	h.maxBodyBytes = 1048576
+	if v := os.Getenv("MAX_BODY_BYTES"); v != "" {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 {
+			h.maxBodyBytes = n
+		}
+	}
+
+	limits := AdmissionLimits{
+		MaxInFlight: admissionMaxInFlight,
+		MaxKVUsage:  admissionMaxKVUsage,
+		MaxPark:     time.Duration(maxParkSeconds) * time.Second,
+	}
+
+	// Issue #64: only backends that actually publish the vLLM scheduler gauges
+	// may be observed. `reasoning` is the only vLLM engine here; ornith and
+	// qwen-cpu are Ollama and embeddings/rerank are TEI, none of which export
+	// vllm:* series. Observing them would trip the startup self-check and hold
+	// observer_degraded at 1 forever. Unset (the default) observes nothing and
+	// leaves admission exactly as it was before this change.
+	observeModels := make(map[string]bool)
+	for _, name := range strings.Split(os.Getenv("ADMISSION_OBSERVE_MODELS"), ",") {
+		if name = strings.TrimSpace(name); name != "" {
+			observeModels[name] = true
+		}
+	}
+
+	h.engineObservers = make(map[string]*EngineObserver)
 	h.schedulers = make(map[string]*Scheduler)
 	for modelName, model := range cfg.Models {
 		addrs := model.AllAddresses()
-		if len(addrs) > 0 {
-			h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue)
+		if len(addrs) == 0 {
+			continue
 		}
+		if observeModels[modelName] {
+			for _, addr := range addrs {
+				if _, ok := h.engineObservers[addr]; !ok {
+					h.engineObservers[addr] = NewEngineObserver(addr, enginePollInterval)
+				}
+			}
+		}
+		h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue, h.engineObservers, limits)
+	}
+	// The handler has no request-path lifecycle to hang cancellation on, so the
+	// observers poll for the lifetime of the process.
+	for _, o := range h.engineObservers {
+		o.Start(context.Background())
 	}
 	// Pre-populate transports for all scheduler backends at startup (single-threaded).
 	// getOrCreateTransport is not goroutine-safe; pre-populating avoids concurrent writes.
@@ -151,7 +239,7 @@ func New(cfg *config.Config) *Handler {
 		// Create a transport per unique upstream address for connection reuse
 		transport := h.getOrCreateTransport(route.Upstream.Address, &route.Upstream)
 		
-		upstreamURL, _ := url.Parse("http://" + route.Upstream.Address)
+		upstreamURL := &url.URL{Scheme: "http", Host: route.Upstream.Address}
 		
 		r := &Route{
 			Name:      name,
@@ -314,6 +402,10 @@ func writeProblemDetail(w http.ResponseWriter, status int, problemType, title, d

 // ServeHTTP implements http.Handler.
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// Captured before any work so TTFT measures from the moment the gateway
+	// accepted the request, not from when the upstream response is copied.
+	requestStart := time.Now()
+
 	if h.handleAuthEndpoint(w, r) {
 		return
 	}
@@ -358,7 +450,14 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// For streaming responses (SSE, chunked), disable buffering. Issue #33.
 	w.Header().Set("X-Accel-Buffering", "no")

-	proxy.ServeHTTP(w, r)
+	// Capture per-request LLM latency metrics (TTFT/ITL/tokens) while the
+	// response is copied from the upstream. Inert for non-SSE responses.
+	var respWriter http.ResponseWriter = w
+	if route.ModelName != "" {
+		respWriter = NewResponseWriterWrapper(w, route.ModelName, observability.Global(), requestStart)
+	}
+
+	proxy.ServeHTTP(respWriter, r)
 }


@@ -384,27 +483,104 @@ func (h *Handler) applyWarmup(ctx context.Context, r *http.Request) bool {
 // acquireScheduledBackend selects a backend from sched, applies warmup
 // deprioritisation, acquires a concurrency slot, and patches route when the
 // selected backend differs from the config default. It returns the backend
-// to Release (caller must defer), the (possibly patched) route, and ok=true.
-// If the slot cannot be acquired it writes a 503 and returns (nil, nil, false).
-func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool) {
+// to Release (caller must defer), the (possibly patched) route, whether SSE
+// headers were already written while parking, and ok=true.
+// If the slot cannot be acquired it writes a 503 and returns ok=false.
+func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool, bool) {
 	backend := sched.Select(cacheKey)
 	if !h.applyWarmup(ctx, r) {
-		return nil, nil, false
+		return nil, nil, false, false
 	}
 	if backend == nil {
-		return nil, route, true
+		return nil, route, false, true
 	}
-	if err := backend.Acquire(ctx); err != nil {
-		if errors.Is(err, ErrQueueFull) {
+
+	// Decision Q22: a request that actually parks on the engine gate and wants
+	// an SSE stream gets its headers pre-written plus a 15 s keep-alive, so no
+	// intermediary times the connection out before the engine admits it.
+	var (
+		prewritten bool
+		hbStop     chan struct{}
+		hbDone     chan struct{}
+	)
+	onPark := func() {
+		if !requestWantsStream(r) {
+			return
+		}
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			return
+		}
+		prewritten = true
+		hbStop = make(chan struct{})
+		hbDone = make(chan struct{})
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("Connection", "keep-alive")
+		w.Header().Set("X-Accel-Buffering", "no")
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprint(w, ": keep-alive\n\n")
+		flusher.Flush()
+		go func() {
+			defer close(hbDone)
+			ticker := time.NewTicker(15 * time.Second)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-hbStop:
+					return
+				case <-ctx.Done():
+					return
+				case <-ticker.C:
+					fmt.Fprint(w, ": keep-alive\n\n")
+					flusher.Flush()
+				}
+			}
+		}()
+	}
+
+	err := backend.acquire(ctx, onPark)
+	if prewritten {
+		close(hbStop)
+		<-hbDone
+	}
+	if err != nil {
+		// Once the SSE status line is spent, report the rejection as an event
+		// instead of a second WriteHeader.
+		reject := func(body string) {
+			if prewritten {
+				fmt.Fprintf(w, "event: error\ndata: {\"error\":%q}\n\n", body)
+				if f, ok := w.(http.Flusher); ok {
+					f.Flush()
+				}
+				return
+			}
+			w.WriteHeader(http.StatusServiceUnavailable)
+			fmt.Fprintf(w, "%s", body)
+		}
+		if errors.Is(err, ErrAdmissionTimeout) {
+			observability.Global().IncrAdmissionRejected("park_timeout")
+			logging.Warnf("admission timeout", map[string]string{
+				"backend": backend.Address,
+				"path":    r.URL.Path,
+				"model":   route.ModelName,
+			})
+			if !prewritten {
+				w.Header().Set("Retry-After", "5")
+			}
+			reject("admission timeout")
+		} else if errors.Is(err, ErrQueueFull) {
 			observability.Global().IncrQueueFull()
+			observability.Global().IncrAdmissionRejected("queue_full")
 			logging.Warnf("backend queue full", map[string]string{
 				"backend": backend.Address,
 				"path":    r.URL.Path,
 				"model":   route.ModelName,
 			})
-			w.Header().Set("Retry-After", "5")
-			w.WriteHeader(http.StatusServiceUnavailable)
-			fmt.Fprintf(w, "queue full")
+			if !prewritten {
+				w.Header().Set("Retry-After", "5")
+			}
+			reject("queue full")
 		} else {
 			logging.Warnf("backend acquire failed", map[string]string{
 				"backend": backend.Address,
@@ -412,15 +588,14 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr
 				"model":   route.ModelName,
 				"err":     err.Error(),
 			})
-			w.WriteHeader(http.StatusServiceUnavailable)
-			fmt.Fprintf(w, "upstream busy")
+			reject("upstream busy")
 		}
-		return nil, nil, false
+		return nil, nil, prewritten, false
 	}
 	if backend.Address != route.Upstream.Address {
 		patchedCfg := *route.Upstream
 		patchedCfg.Address = backend.Address
-		patchedURL, _ := url.Parse("http://" + backend.Address)
+		patchedURL := &url.URL{Scheme: "http", Host: backend.Address}
 		route = &Route{
 			Name:      route.Name,
 			ModelName: route.ModelName,
@@ -431,7 +606,27 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr
 			},
 		}
 	}
-	return backend, route, true
+	return backend, route, prewritten, true
+}
+
+// requestWantsStream reports whether the JSON request body asks for an SSE
+// stream ("stream": true). The body is restored for the upstream.
+func requestWantsStream(r *http.Request) bool {
+	if r.Body == nil {
+		return false
+	}
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		return false
+	}
+	r.Body = io.NopCloser(bytes.NewReader(body))
+	var probe struct {
+		Stream bool `json:"stream"`
+	}
+	if err := json.Unmarshal(body, &probe); err != nil {
+		return false
+	}
+	return probe.Stream
 }

 // handleAuthEndpoint routes /auth/* requests and returns true if handled.
@@ -548,16 +743,17 @@ func (h *Handler) authenticateV1(w http.ResponseWriter, r *http.Request) bool {
 		return false
 	}
 	identity.Inject(r, claims)
-	if h.config.Auth.RequiredCapability != "" {
-		if !h.jwtValidator.CheckPermissions(claims, h.config.Auth.RequiredCapability, "*") {
+	if caps := requiredCapabilities(h.config.Auth.RequiredCapability); len(caps) > 0 {
+		args := append(caps, "*")
+		if !h.jwtValidator.CheckPermissions(claims, args...) {
 			writeProblemDetail(w, http.StatusForbidden,
 				"https://api.example.com/problems/insufficient-permissions",
 				"Insufficient Permissions",
-				fmt.Sprintf("Required capability: %s", h.config.Auth.RequiredCapability),
+				fmt.Sprintf("Required capability: %s", strings.Join(caps, " or ")),
 				nil)
 			logging.Errorf("auth failed", fmt.Errorf("insufficient permissions"), map[string]string{
 				"path":     r.URL.Path,
-				"required": h.config.Auth.RequiredCapability,
+				"required": strings.Join(caps, ","),
 			})
 			return false
 		}
@@ -565,6 +761,22 @@ func (h *Handler) authenticateV1(w http.ResponseWriter, r *http.Request) bool {
 	return true
 }

+// requiredCapabilities splits the comma-separated requiredCapability setting
+// into a list. Several spellings are accepted because the same capability is
+// named differently by each identity provider: Authentik used "llm:inference"
+// as a permission string, while ZITADEL uses role keys such as
+// "llm-inference". A token satisfies the gate when it carries any one of them.
+func requiredCapabilities(raw string) []string {
+	parts := strings.Split(raw, ",")
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if p = strings.TrimSpace(p); p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
 // enforceBodyLimit caps request body size for non-chat routes.
 // Returns false when the request is rejected (response already written).
 func (h *Handler) enforceBodyLimit(w http.ResponseWriter, r *http.Request, route *Route) bool {
@@ -594,9 +806,10 @@ func (h *Handler) enforceBodyLimit(w http.ResponseWriter, r *http.Request, route
 // Returns (nil, nil, false) when the backend slot cannot be acquired (response written).
 func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *http.Request, route *Route, sched *Scheduler, cacheKey string) (*httputil.ReverseProxy, *Backend, bool) {
 	var selectedBackend *Backend
+	var headersPrewritten bool
 	if sched != nil {
 		var ok bool
-		selectedBackend, route, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey)
+		selectedBackend, route, headersPrewritten, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey)
 		if !ok {
 			return nil, nil, false
 		}
@@ -611,12 +824,58 @@ func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *h

 	capturedBackend := selectedBackend
 	capturedRoute := route
+	if headersPrewritten {
+		// The 200 + text/event-stream status is already on the wire, so an
+		// upstream status can no longer be forwarded: httputil.ReverseProxy
+		// calls WriteHeader again and net/http keeps the first one. Frame a
+		// failure as an SSE event rather than letting a raw error body land
+		// inside the stream, which a streaming client reads as an empty
+		// completion and reports as success.
+		proxy.ModifyResponse = func(resp *http.Response) error {
+			if resp.StatusCode < http.StatusBadRequest {
+				return nil
+			}
+			body, readErr := io.ReadAll(io.LimitReader(resp.Body, 4096))
+			resp.Body.Close()
+			if readErr != nil {
+				logging.Warnf("reading upstream error body failed", map[string]string{
+					"upstream": capturedRoute.Upstream.Address,
+					"err":      readErr.Error(),
+				})
+			}
+			logging.Warnf("upstream error after parked stream started", map[string]string{
+				"upstream": capturedRoute.Upstream.Address,
+				"status":   strconv.Itoa(resp.StatusCode),
+			})
+			resp.Body = io.NopCloser(strings.NewReader(fmt.Sprintf(
+				"event: error\ndata: {\"upstream_status\":%d,\"error\":%q}\n\n",
+				resp.StatusCode, strings.TrimSpace(string(body)))))
+			resp.ContentLength = -1
+			resp.Header.Del("Content-Length")
+			resp.Header.Set("Content-Type", "text/event-stream")
+			return nil
+		}
+	}
 	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
 		if capturedBackend != nil {
 			if s := h.schedulers[capturedRoute.ModelName]; s != nil {
 				s.MarkFailed(capturedBackend.Address)
 			}
 		}
+		if headersPrewritten {
+			// SSE headers and a 200 are already on the wire: a 502 here would be
+			// an ignored WriteHeader and would tear the event stream. Report the
+			// failure as an event instead.
+			fmt.Fprint(w, "event: error\ndata: {\"error\":\"upstream error\"}\n\n")
+			if f, ok := w.(http.Flusher); ok {
+				f.Flush()
+			}
+			logging.Errorf("upstream error", err, map[string]string{
+				"upstream": capturedRoute.Upstream.Address,
+				"path":     r.URL.Path,
+			})
+			return
+		}
 		w.WriteHeader(http.StatusBadGateway)
 		fmt.Fprintf(w, "upstream error")
 		logging.Errorf("upstream error", err, map[string]string{
@@ -69,9 +69,10 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	// Restore the body so it can be read again by the upstream
 	r.Body = io.NopCloser(bytes.NewReader(bodyBytes))

-	// Parse the JSON to find the model field
-	var payload map[string]interface{}
-	if err := json.Unmarshal(bodyBytes, &payload); err != nil {
+	// Extract only the top-level "model" value with a token walk: the common
+	// path must not materialise the whole body (messages and all) as a map.
+	modelRaw, hasModel, err := extractTopLevelModel(bodyBytes)
+	if err != nil {
 		return nil, &modelValidationError{
 			Kind:    "invalid_json",
 			Message: "request body is not valid JSON",
@@ -79,7 +80,6 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	}

 	// Extract the model name
-	modelVal, hasModel := payload["model"]
 	if !hasModel {
 		return nil, &modelValidationError{
 			Kind:    "missing_model",
@@ -88,7 +88,7 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	}

 	// Handle null model
-	if modelVal == nil {
+	if bytes.Equal(bytes.TrimSpace(modelRaw), []byte("null")) {
 		return nil, &modelValidationError{
 			Kind:    "null_model",
 			Message: "'model' field is null",
@@ -96,8 +96,8 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	}

 	// Extract as string
-	modelName, ok := modelVal.(string)
-	if !ok {
+	var modelName string
+	if err := json.Unmarshal(modelRaw, &modelName); err != nil {
 		return nil, &modelValidationError{
 			Kind:    "missing_model",
 			Message: "'model' field must be a string",
@@ -122,10 +122,26 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 		}
 	}

-	// If upstream expects a different model name, rewrite the body
+	// If upstream expects a different model name, rewrite the body. This path is
+	// uncommon, so it may decode the body as raw fields; the common path above
+	// does not build a full map.
 	if modelUpstream.UpstreamModel != "" && modelUpstream.UpstreamModel != modelName {
-		payload["model"] = modelUpstream.UpstreamModel
-		newBody, err := json.Marshal(payload)
+		var fields map[string]json.RawMessage
+		if err := json.Unmarshal(bodyBytes, &fields); err != nil {
+			return nil, &modelValidationError{
+				Kind:    "invalid_request",
+				Message: fmt.Sprintf("failed to rewrite model name: %v", err),
+			}
+		}
+		newModel, err := json.Marshal(modelUpstream.UpstreamModel)
+		if err != nil {
+			return nil, &modelValidationError{
+				Kind:    "invalid_request",
+				Message: fmt.Sprintf("failed to rewrite model name: %v", err),
+			}
+		}
+		fields["model"] = newModel
+		newBody, err := json.Marshal(fields)
 		if err != nil {
 			return nil, &modelValidationError{
 				Kind:    "invalid_request",
@@ -151,11 +167,11 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 		ConnectTimeout: h.defaultConnectTimeout,
 		ReadTimeout:    h.defaultReadTimeout,
 		WriteTimeout:   h.defaultWriteTimeout,
-		MaxBodySize:    h.defaultMaxBodySize,
+		MaxBodySize:    h.maxBodyBytes,
 		AuthRequired:   false,
 	}

-	targetURL, _ := url.Parse("http://" + modelUpstream.Address)
+	targetURL := &url.URL{Scheme: "http", Host: modelUpstream.Address}

 	route := &Route{
 		Name:      "v1-chat-" + modelName,
@@ -169,3 +185,87 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {

 	return route, nil
 }
+
+// extractTopLevelModel walks a JSON object body with a streaming decoder and
+// returns the raw value of the top-level "model" field. It never builds the
+// full document as a map, so a large "messages" array is skipped by token.
+// hasModel is false when the key is absent or the document is a bare null.
+func extractTopLevelModel(body []byte) (model json.RawMessage, hasModel bool, err error) {
+	dec := json.NewDecoder(bytes.NewReader(body))
+	tok, err := dec.Token()
+	if err != nil {
+		return nil, false, err
+	}
+	if tok == nil {
+		// A bare `null` body decoded to a nil map before, i.e. no model field.
+		if _, err := dec.Token(); err != io.EOF {
+			if err == nil {
+				return nil, false, fmt.Errorf("unexpected data after JSON null")
+			}
+			return nil, false, err
+		}
+		return nil, false, nil
+	}
+	if delim, ok := tok.(json.Delim); !ok || delim != '{' {
+		return nil, false, fmt.Errorf("request body is not a JSON object")
+	}
+	for dec.More() {
+		keyTok, err := dec.Token()
+		if err != nil {
+			return nil, false, err
+		}
+		key, _ := keyTok.(string)
+		if key == "model" {
+			if err := dec.Decode(&model); err != nil {
+				return nil, false, err
+			}
+			hasModel = true
+			continue
+		}
+		if err := skipJSONValue(dec); err != nil {
+			return nil, false, err
+		}
+	}
+	// Consume the closing brace and reject trailing data, preserving the
+	// invalid_json answer the old whole-body decode produced.
+	if _, err := dec.Token(); err != nil {
+		return nil, false, err
+	}
+	if _, err := dec.Token(); err != io.EOF {
+		if err == nil {
+			return nil, false, fmt.Errorf("unexpected data after JSON object")
+		}
+		return nil, false, err
+	}
+	return model, hasModel, nil
+}
+
+// skipJSONValue advances dec past the next JSON value without materialising it.
+func skipJSONValue(dec *json.Decoder) error {
+	tok, err := dec.Token()
+	if err != nil {
+		return err
+	}
+	delim, ok := tok.(json.Delim)
+	if !ok {
+		return nil // scalar
+	}
+	if delim != '{' && delim != '[' {
+		return fmt.Errorf("unexpected delimiter %v", delim)
+	}
+	depth := 1
+	for depth > 0 {
+		tok, err := dec.Token()
+		if err != nil {
+			return err
+		}
+		if d, ok := tok.(json.Delim); ok {
+			if d == '{' || d == '[' {
+				depth++
+			} else {
+				depth--
+			}
+		}
+	}
+	return nil
+}
@@ -8,20 +8,41 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
 )

 // ErrQueueFull is returned when the in-memory queue is at capacity.
 var ErrQueueFull = errors.New("backend queue full")

+// ErrAdmissionTimeout is returned when a request parked on engine admission
+// exceeds the configured maximum park duration.
+var ErrAdmissionTimeout = errors.New("admission timeout")
+
+// AdmissionLimits are the engine-wide admission thresholds applied to every
+// backend of a scheduler (#64).
+type AdmissionLimits struct {
+	MaxInFlight int
+	MaxKVUsage  float64
+	MaxPark     time.Duration
+}
+
 // Backend represents a single vLLM upstream with a bounded in-memory queue.
 // Two channels model the queue:
 //
 //	queue  — waiting requests (bounded by VLLM_MAX_QUEUE_DEPTH)
 //	slots  — concurrent in-service requests (bounded by VLLM_MAX_NUM_SEQS)
+//
+// The engine gate is applied before the queue: admission reads the engine's
+// own published gauges via observer, which is global across replicas.
 type Backend struct {
-	Address string
-	queue   chan struct{}
-	slots   chan struct{}
+	Address     string
+	queue       chan struct{}
+	slots       chan struct{}
+	observer    *EngineObserver
+	maxInFlight int
+	maxKVUsage  float64
+	maxPark     time.Duration
 }

 // newBackend creates a Backend with given queue depth and concurrency limit.
@@ -40,7 +61,48 @@ func newBackend(addr string, maxSlots, maxQueue int) *Backend {
 // Acquire enters the in-memory queue then waits for a vLLM concurrency slot.
 // Returns ErrQueueFull immediately if the queue channel is at capacity (no goroutine queued).
 // Returns ctx.Err() if the context is cancelled while waiting for a slot.
+// Returns ErrAdmissionTimeout if the engine gate parks the request past maxPark.
 func (b *Backend) Acquire(ctx context.Context) error {
+	return b.acquire(ctx, nil)
+}
+
+// acquire is Acquire with an optional onPark callback. onPark is invoked once,
+// on the calling goroutine, the first time the request has to wait for engine
+// admission; the parked streaming path uses it to start an SSE heartbeat.
+func (b *Backend) acquire(ctx context.Context, onPark func()) error {
+	// Engine gate (#64): admit on the engine's own gauges, not on this pod's
+	// counters. Failing OPEN on !Healthy is deliberate — a metrics outage must
+	// not become an inference outage; the three-replica gate is only a fairness
+	// improvement on top of the engine's own queueing.
+	if b.observer != nil {
+		deadline := time.Now().Add(b.maxPark)
+		timer := time.NewTimer(time.Until(deadline))
+		defer timer.Stop()
+		parked := false
+		for {
+			// State and tick channel are read together under one lock: taking
+			// them separately can miss a poll that lands between the two calls.
+			st, tick := b.observer.Snapshot()
+			if !st.Healthy || (st.InFlight() < int64(b.maxInFlight) && st.KVUsage < b.maxKVUsage) {
+				break
+			}
+			if !parked {
+				observability.Global().IncrAdmissionParked()
+				parked = true
+				if onPark != nil {
+					onPark()
+				}
+			}
+			select {
+			case <-tick:
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-timer.C:
+				return ErrAdmissionTimeout
+			}
+		}
+	}
+
 	// Non-blocking queue entry: reject immediately when queue is full.
 	select {
 	case b.queue <- struct{}{}:
@@ -76,10 +138,15 @@ type Scheduler struct {

 // NewScheduler creates a Scheduler for the given addresses.
 // maxSlots = VLLM_MAX_NUM_SEQS, maxQueue = VLLM_MAX_QUEUE_DEPTH.
-func NewScheduler(addresses []string, maxSlots, maxQueue int) *Scheduler {
+// observers holds one engine observer per backend address (may be nil).
+func NewScheduler(addresses []string, maxSlots, maxQueue int, observers map[string]*EngineObserver, limits AdmissionLimits) *Scheduler {
 	backends := make([]*Backend, len(addresses))
 	for i, addr := range addresses {
 		backends[i] = newBackend(addr, maxSlots, maxQueue)
+		backends[i].observer = observers[addr]
+		backends[i].maxInFlight = limits.MaxInFlight
+		backends[i].maxKVUsage = limits.MaxKVUsage
+		backends[i].maxPark = limits.MaxPark
 	}
 	return &Scheduler{
 		backends:    backends,
@@ -67,7 +67,7 @@ func TestBackendAcquireReturnsErrQueueFullWhenQueueFull(t *testing.T) {
 // always selects the same backend (rendezvous / highest-random-weight hashing).
 func TestSchedulerSelectRendezvousHashDeterministic(t *testing.T) {
 	addrs := []string{"a:80", "b:80", "c:80"}
-	s := NewScheduler(addrs, 10, 10)
+	s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{})

 	first := s.Select("same-cache-key")
 	if first == nil {
@@ -85,7 +85,7 @@ func TestSchedulerSelectRendezvousHashDeterministic(t *testing.T) {
 // requests evenly across all backends when no cache key is given.
 func TestSchedulerSelectNoKeyRoundRobin(t *testing.T) {
 	addrs := []string{"a:80", "b:80", "c:80"}
-	s := NewScheduler(addrs, 10, 10)
+	s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{})

 	counts := make(map[string]int)
 	for i := 0; i < 9; i++ {
@@ -160,7 +160,7 @@ func TestBackendAcquireCancelledContext(t *testing.T) {
 // selected within the fail window (30 s default).
 func TestSchedulerMarkFailedExcludesBackend(t *testing.T) {
 	addrs := []string{"a:80", "b:80"}
-	s := NewScheduler(addrs, 10, 10)
+	s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{})

 	s.MarkFailed("a:80")

@@ -36,6 +36,21 @@ func (rw *responseWriter) Write(b []byte) (int, error) {
 	return n, err
 }

+// Flush forwards to the wrapped writer so that streaming handlers can still
+// detect http.Flusher through this wrapper. Without it a bare
+// `w.(http.Flusher)` assertion fails for every request that passes through
+// this middleware, which silently disables SSE flushing and keep-alives.
+func (rw *responseWriter) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+// Unwrap exposes the wrapped writer to http.ResponseController.
+func (rw *responseWriter) Unwrap() http.ResponseWriter {
+	return rw.ResponseWriter
+}
+
 // Middleware returns an HTTP middleware that adds tracing to requests.
 func Middleware(next http.Handler) http.Handler {
 	tracer := otel.Tracer(tracerName)
@@ -21,9 +21,10 @@ spec:
      labels:
        app: api-gateway
        component: gateway
-        # Required by the llm-serving-default-deny NetworkPolicy, which admits
-        # only pods labelled llm-client=true (from any namespace) on port 8080.
-        # Without it every upstream dial times out and dispatch returns 502.
+        # The llm-serving-ingress NetworkPolicy in the llm-serving namespace now
+        # selects by namespace (kubernetes.io/metadata.name: api), not by this
+        # label. Kept for older NetworkPolicy revisions that still admit on
+        # llm-client=true.
        llm-client: "true"
      annotations:
        reloader.stakater.com/auto: "true"
@@ -67,6 +68,24 @@ spec:
          value: "5m"
        - name: LOG_LEVEL
          value: "info"
+        - name: ADMISSION_MAX_PARK_SECONDS
+          value: "120"
+        # Must track the engine's --max-num-seqs (homelab: reasoning.yaml).
+        # The gauge this compares against is engine-wide, so this number is a
+        # ceiling on concurrent requests across all gateway replicas, not a
+        # per-pod slot count. Drift here is silent; startup logs the value.
+        - name: ADMISSION_MAX_INFLIGHT
+          value: "4"
+        # Only backends that publish the vLLM scheduler gauges may be observed.
+        # `reasoning` is the only vLLM engine; ornith/qwen-cpu are Ollama and
+        # embeddings/rerank are TEI, so observing them would hold
+        # observer_degraded at 1 forever.
+        - name: ADMISSION_OBSERVE_MODELS
+          value: "reasoning"
+        - name: VLLM_MAX_QUEUE_DEPTH
+          value: "8"
+        - name: MAX_BODY_BYTES
+          value: "1048576"
        # OpenTelemetry tracing configuration
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "otel-collector.tracing.svc.cluster.local:4317"
@@ -21,11 +21,17 @@ spec:
    ports:
    - protocol: TCP
      port: 8080
-  # Allow from Prometheus scraping (if in monitoring namespace)
+  # Allow from Prometheus scraping (monitoring namespace)
+  # MUST select on kubernetes.io/metadata.name, not a plain `name:` label.
+  # Only ingress-nginx carries a plain `name` label in this cluster; every
+  # other namespace has just kubernetes.io/metadata.name, so `name: monitoring`
+  # matched nothing and silently denied every Prometheus scrape of :8080
+  # (surfaced as context-deadline-exceeded on all 3 gateway replicas while the
+  # ServiceMonitor itself was discovered and healthy).
  - from:
    - namespaceSelector:
        matchLabels:
-          name: monitoring
+          kubernetes.io/metadata.name: monitoring
    ports:
    - protocol: TCP
      port: 8080
@@ -93,19 +99,21 @@ spec:
    ports:
    - protocol: TCP
      port: 8080
-  # Allow to IAM (Authentik) for JWKS fetch + API
-  # Authentik pod listens on 9000 (http) and 9443 (https)
-  # Service translates 80→9000, 443→9443
-  # NetworkPolicy matches destination pod port, not service port
+  # Allow to IAM (Zitadel) for JWKS fetch + API
+  # Zitadel API listens on 8080 (h2c/plaintext in-cluster); the login UI
+  # listens on 3000. Public access goes through ingress-nginx on 443 (already
+  # covered by the 10.96.0.0/12 CIDR below), but 8080 keeps direct in-cluster
+  # calls working if CoreDNS rewrite is bypassed.
+  # NetworkPolicy matches destination pod port, not service port.
  - to:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/metadata.name: iam
    ports:
    - protocol: TCP
-      port: 9000
+      port: 8080
    - protocol: TCP
-      port: 9443
+      port: 3000
  # Allow to SQS (queue management service)
  - to:
    - namespaceSelector:
Author	SHA1	Message	Date
Admin Bot	ba8800d565	fix(gateway): set jwksUrl so the dispatcher builds a validator CI / CI (pull_request) Successful in 2m9s Details CI / CI (push) Successful in 3m29s Details cmd/gateway/main.go:144 constructs the adapter validator only when both cfg.Auth.Enabled and cfg.Auth.JWKSURL are set. The config left jwksUrl unset on purpose, so jwtValidator stayed nil and every adapter request passed with one warning line: auth required but jwtValidator nil; request unauthenticated That made auth.required cosmetic: sqs, workflow, memory, s3 and iam all answered unauthenticated calls. Set the JWKS URL to the ZITADEL key endpoint.	2026-10-27 01:17:09 -07:00
Admin Bot	29f5c64fab	feat(gateway): require a token on every adapter CI / CI (pull_request) Successful in 3m0s Details CI / CI (push) Successful in 3m26s Details workflow, memory, s3 and iam served requests without any Authorization header. Only sqs and the LLM models checked a token, so four upstreams were reachable from inside the cluster with no authentication at all. Set auth.required true on all five adapters. The dispatcher then enforces one capability per method, derived from the verb (internal/serviceadapter/router.go:142): GET/HEAD/OPTIONS -> "<service>:read" anything else -> "<service>:write" Those strings must exist as ZITADEL role keys with the colon. The project now holds memory:read, memory:write, workflow:read, workflow:execute, s3:read, s3:write, iam:read, iam:write, sqs:read and sqs:write, and the riotpiao user grant carries them, so a browser session keeps working. Correct the old comment: ZITADEL does store colon role keys. Verified by listing the project roles, which returned all eleven colon keys.	2026-10-27 01:11:18 -07:00
Admin Bot	8875f36dde	feat(auth): validate access tokens with the ZITADEL Go SDK Replace the handwritten Authentik JWT checks with github.com/zitadel/zitadel-go/v3. What the SDK now owns: - OIDC discovery (client.Discover) to find issuer and JWKS URI - remote, auto-refreshing JWKS (rp.NewRemoteKeySet) - signature, expiry and issuer checks (op.NewAccessTokenVerifier, op.VerifyAccessToken) - the verified claim set and role lookup (oauth.IntrospectionContext, IsGrantedRole) What stays local: - a multi-audience allow-list, because this gateway is a shared resource server and the SDK's oauth.WithJWT accepts exactly one client ID - an adapter from IntrospectionContext to jwt.MapClaims so the rest of the gateway keeps one claim representation Other changes: - issuer is configuration-driven; the hardcoded authentik.riotpiao.com check is gone - aud may be a string or an array (ZITADEL sends an array) - CheckPermissions also reads urn:zitadel:iam:org:project:roles, the claim ZITADEL uses for project roles - requiredCapability accepts a comma-separated list, so llm:inference and llm-inference both satisfy the LLM gate - a five-segment token is reported as an encrypted JWE with the fix in the message, instead of failing as a bad signature Config: auth.zitadelDomain turns the SDK path on; jwksUrl stays supported for the legacy path and for tests.	2026-10-27 01:11:18 -07:00
poimen rock Poimen	727b9e1cb0	fix(proxy): admission-control review fixes missing from the #65 merge (#66 ) CI / CI (push) Successful in 43m8s Details Follow-up to #65, which is already merged. The review of that PR returned BLOCK with two high-severity defects; the fix commit was pushed after the merge landed, so none of it is in `main`. ## What happened - `main` @ `369d51a` is the squash of the first commit only (`c5a91cf`). - `d71c884` (review fixes) is on the old branch and was never merged — PR #65 is closed. - Verified against `main` directly: `ADMISSION_MAX_INFLIGHT` 0 hits, `func (rw responseWriter) Flush` 0 hits, `ModifyResponse` 0 hits, `Snapshot` 0 hits. - The image in production (`forgejo.riotpiao.com/rock/api-gateway@sha256:64f2bfc1…`, 3/3 replicas) is built from that squash, so the defects below are live right now. ## Live state confirmed `https://api.riotpiao.com/metrics` on the running pods: ``` observer_degraded 0 vllm_requests_running 1 vllm_kv_usage_perc 0.008293092588704587 admission_parked_total 0 engine_state_stale_total 0 ``` The observer itself works: it is reading the real vLLM exposition body and the startup self-check passed (`observer_degraded 0`). No requests have parked yet, so the two defects below have not been observable in the wild — they are proven from code, not from a live failure. ## Defects fixed here 1. The parked-stream keep-alive never runs in production (high).* `internal/tracing/middleware.go` wraps the `ResponseWriter` in a struct that embeds the interface and implements only `WriteHeader`/`Write`, so `w.(http.Flusher)` in `internal/proxy/proxy.go` is false for every real request (`cmd/gateway/main.go` → `tracing.Middleware` → `router`). `prewritten` therefore never becomes true, the heartbeat never starts, and a parked streaming request writes zero bytes for up to `ADMISSION_MAX_PARK_SECONDS` (120) — the client timeout the feature exists to prevent. Fixed by adding `Flush()` and `Unwrap()` to the wrapper. 2. Upstream status masked once the SSE status is spent (high, latent behind 1). `httputil.ReverseProxy` calls `rw.WriteHeader(res.StatusCode)` unconditionally; after the park wrote 200, net/http keeps 200 and the upstream error body is appended raw into the event stream, which a streaming client reads as an empty completion and reports as success. Fixed with a `ModifyResponse` that frames an upstream `>= 400` as an `event: error` when the status has already been written. 3. Lost wakeup in the park loop. `Get()` and `Tick()` were taken under two separate lock acquisitions, so a poll landing between them pushed the waiter onto the next generation's channel — up to one extra poll interval (2s) parked. `Snapshot()` now returns state and channel under one lock. 4. `observer_degraded` pinned at 1 forever. It was raised only on the first poll's parse failure and `SetObserverDegraded(false)` had no call site, so one truncated response pinned the gauge for the pod's life, while genuine metric-name drift appearing later was never reported at all. `markDegraded` now tracks current parsability in both directions and only touches the exporter on a change. 5. `ADMISSION_MAX_INFLIGHT` silently defaulted. Now set explicitly to `4` in `k8s/deployment.yaml` with the coupling to the engine's `--max-num-seqs` documented, instead of quietly inheriting `VLLM_MAX_NUM_SEQS`. ## Verification `go build ./...` clean · `go vet ./...` clean · `go test ./...` 368 passed, 0 failed · `structural-check.sh` 0 errors (1 pre-existing `:latest` warning) · `go.mod` unchanged. Cherry-picked onto `main` rather than merged from the old branch, so this diff is exactly the five fixes and nothing else: 5 files, +80/−8. ## Still open The reviewer's BLOCK also rested on nine functions over CRAP>30 purely from 0% coverage — the deliberate "no new tests" decision. Worth recording that defect 1 is a one-line type assertion that a `httptest` case through the real middleware chain would have caught immediately. Raising it here so the decision is visible rather than implied. --------- Co-authored-by: Poimen <[email protected]> Reviewed-on: #66 Co-authored-by: poimen <[email protected]>	2026-09-30 12:19:24 +00:00
poimenandrock	369d51a5c2	feat(proxy): engine-wide LLM admission control via vLLM gauges (#65 ) CI / CI (push) Successful in 41m20s Details Closes #64 Replaces the per-pod concurrency semaphore as the effective cap with admission gated on the vLLM engine's own scheduler gauges. ## Why `internal/proxy/proxy.go:119` built a semaphore of 4 slots inside each gateway process, and `api-gateway` runs 3 replicas, so the advertised cap was 12 while the engine's `--max-num-seqs` is 4. Nothing coordinated replicas, and the mismatch was invisible because the queue never rejected anything (`queue_full_requests_total 0`). Measured on the engine while investigating a 10-minute TTFT: - `vllm:cache_config_info{num_gpu_blocks="9527",block_size="16"}` = 152,432 tokens of KV capacity for the whole engine. PP=2 splits layers, so both ranks hold KV for the same tokens — not 2x that. - Uncached prefill runs at 262 tok/s contended, 525 tok/s over 7d (`vllm:request_prefill_time_seconds_sum/count`). - One request showed `llm_ttft_seconds_sum 590.981 / count 5` at the gateway and a `(480, 960]` s entry in `vllm:request_prefill_time_seconds_bucket`, with engine-side generation throughput at 0.0-0.1 tok/s throughout: a 65k-token prompt costs ~250 s of prefill and starves every other request while it runs. - `vllm:num_preemptions_total = 0`; KV peaked at `0.5399`. Reading the engine's own gauges is global by construction, needs no lock and no new datastore, and it also accounts for callers that never pass the gateway — namespace `poimen` is admitted directly on 80/8080 by `llm-serving-ingress`, and `memory/crates/mem-cli/src/kv_warmer.rs` is such a caller. A distributed lock would have been blind to that traffic. ## Changes - new `internal/proxy/engine_state.go` — 2s scrape of `http://<addr>/metrics` on its own client (the proxy transports carry socket options tuned for inference), hand-rolled exposition parse, `Healthy=false` after 3 consecutive failed polls with `engine_state_stale_total` counted once at the transition, and a startup self-check that raises `observer_degraded` and logs at error level instead of degrading to fail-open silently. - admission gate (`internal/proxy/scheduler.go`) — parks while `kv_cache_usage_perc >= 0.90` or `running + waiting >= ADMISSION_MAX_INFLIGHT`, bounded by `ADMISSION_MAX_PARK_SECONDS`, then `503 + Retry-After: 5`. Unknown or unhealthy state fails open on purpose: a metrics outage must not become an inference outage. - parked streaming heartbeat (`internal/proxy/proxy.go`) — `stream: true` requests get 200 + `text/event-stream` and a `: keep-alive` comment every 15s while parked, so a client idle timeout does not kill a request that is legitimately waiting. Non-streaming requests park silently. - `internal/proxy/router.go` — extract only the top-level `model` field with a `json.Decoder` token walk instead of unmarshalling the whole body into `map[string]interface{}` (that built the entire `messages`+`tools` tree per request, ~300 KB for a 65k-token prompt). `MaxBodySize` now comes from `MAX_BODY_BYTES`; the model path previously inherited the 100 MB default at `proxy.go:93` against a 512Mi container limit. - `cmd/gateway/main.go` — `GOMAXPROCS` from the cgroup CPU quota (cgroup v2 `cpu.max`, v1 fallback). The container is limited to `500m` while Go defaulted to the host core count and was being CFS-throttled. - `k8s/deployment.yaml` — `ADMISSION_MAX_PARK_SECONDS=120`, `ADMISSION_OBSERVE_MODELS=reasoning`, `VLLM_MAX_QUEUE_DEPTH=8`, `MAX_BODY_BYTES=1048576`; corrected the `llm-client` comment (the NetworkPolicy selects by namespace now, not by that label). - deleted `k8s/configmap.yaml` — absent from `k8s/kustomization.yaml`, and the live ConfigMap of that name carries an unrelated key (`gw-config-fixed.yaml`) plus a stale routing table. - `internal/observability/` — `admission_parked_total`, `admission_rejected_total{reason}`, `engine_state_stale_total`, and gauges `vllm_requests_running` / `vllm_kv_usage_perc` (the gateway's own reading, deliberately mirroring the vLLM series so a disagreement is visible) and `observer_degraded`. Names follow the existing convention, which carries no prefix. Out of scope here, in `riotpiao-poimen/homelab#51`: the engine flags (`--max-num-partial-prefills`, `--max-long-partial-prefills`, `--long-prefill-token-threshold`, `--max-num-batched-tokens`, `--max-num-seqs`, `--gpu-memory-utilization`) and the four dashboard panels in `generate-dashboards.py`. ## Verification - `go build ./...` clean, `go vet ./...` clean. - `go test ./...` — 368 tests, 0 failures (baseline on `main` is 368/0). - `structural-check.sh` on the changed files — 0 errors, 1 pre-existing warning (`:latest` image tag). - `go.mod` / `go.sum` byte-identical: no new dependency. Not yet verified: the real-engine load test (2x65k + 15x8k against `reasoning`). That run also produces the number for `ADMISSION_MAX_PARK_SECONDS`, which today is the provisional 120; retuning it is a follow-up since it is env-only and needs no rebuild. ## Accepted risks - No tests were added for the observer or the gate (deliberate: production is the test bed). The mitigation is the startup self-check, which is production self-diagnosis rather than a test, plus the `observer_degraded` alert. A parse failure that outlives the first poll logs and surfaces as a metric; it does not silently disable gating without a signal. - The cap is soft, not hard. Between polls each replica may admit more than one request, so there is no clean bound. Exposure is low (~11,500 requests over 7d ≈ 1.4/min) and the engine remains the hard limit — a race parks requests in vLLM's queue rather than over-committing KV. - `ADMISSION_MAX_INFLIGHT` is manually coupled to `--max-num-seqs` and can drift silently. It defaults to `VLLM_MAX_NUM_SEQS`, and startup logs the effective value. - The first scrape waits one interval, so the first ~2s of admission is ungated (fail-open). Scraping at `Start()` instead would make the gateway contact every observed backend before any request needs one, which is wasted work and a surprising side effect. ## Rollback Revert this commit and redeploy. The admission path is fail-open, so the code can also be neutralised without reverting by unsetting `ADMISSION_OBSERVE_MODELS`. Separately, the orphaned cluster object can be removed by hand: `kubectl delete cm -n api api-gateway-config` (not a code change, deliberately left to a human). --------- Reviewed-on: #65 Co-authored-by: poimen <[email protected]>	2026-09-30 10:56:10 +00:00
rock	aac68386d8	fix(proxy): wire LLM latency metrics into the streaming response path (#63 ) CI / CI (push) Successful in 34m13s Details `NewLLMMetricsCapture`/`ResponseWriterWrapper` had zero call sites, so `llm_ttft_seconds`, `llm_itl_seconds` and `llm_tokens_total` were exported as HELP/TYPE stubs with no samples (verified against the live `/metrics`). These are the only per-request LLM latency series covering the Ollama models (`ornith:35b`, `qwen2.5:3b-instruct`), which expose no `/metrics` of their own — so those models had zero latency visibility. ## Change Wire the wrapper around the response writer in `ServeHTTP` for model-routed requests, and delete the unused `io.WriteCloser` duplicate that made the gap hard to spot (two near-identical implementations, neither constructed). ## Correctness details - Only SSE responses are measured. A non-streaming completion is written in one `Write`, which would otherwise be recorded as a TTFT sample equal to the whole request latency, with no ITL samples — plausible-looking but meaningless numbers. - `requestStart` is captured at the top of `ServeHTTP` so TTFT measures from request acceptance, not from when the upstream response is copied. - Token count skips the `data: [DONE]` terminator. - `Unwrap()` keeps `http.ResponseController` (Flush/Hijack/deadlines) working through the wrapper, so SSE unbuffering (#33) is unaffected. ## Tests \| Test \| Guards \| \|---\|---\| \| `TestResponseWriterWrapperRecordsSSEMetrics` \| TTFT=1, ITL>0, tokens=2 (terminator excluded) \| \| `TestResponseWriterWrapperIgnoresNonStreaming` \| JSON response records nothing \| \| `TestResponseWriterWrapperPreservesStreaming` \| bytes + Flush pass through \| \| `TestHandlerRecordsLLMMetricsForStreaming` \| end-to-end through `ServeHTTP` \| The last one fails if the wrapper is ever wired nowhere again — verified by disabling the wiring locally and watching it fail. ## Verification needed post-merge Confirm samples appear: ``` curl -s https://api.riotpiao.com/metrics \| grep -E "llm_(ttft\|itl)_seconds_count\|llm_tokens_total" ```	2026-09-29 15:46:39 +00:00
rockandAdmin Bot	28753d11ed	fix(netpol): allow Prometheus scrape of gateway /metrics (#62 ) CI / CI (push) Successful in 32m48s Details ## Problem The new `api-gateway` ServiceMonitor discovers all 3 replicas, but every scrape fails with `context deadline exceeded`: ``` down api-gateway api-gateway-6986b7c889-pmt4b err=Get "http://10.244.0.180:8080/metrics": context deadline down api-gateway api-gateway-6986b7c889-c28dt err=Get "http://10.244.1.1:8080/metrics": context deadline down api-gateway api-gateway-6986b7c889-552gl err=Get "http://10.244.2.61:8080/metrics": context deadline ``` ## Cause `network-policy.yaml` allowed Prometheus ingress with: ```yaml namespaceSelector: matchLabels: name: monitoring ``` Only `ingress-nginx` carries a plain `name:` label in this cluster. Every other namespace has just `kubernetes.io/metadata.name`, so this rule matched nothing and silently denied the scrape. (The rule directly above it — `name: ingress-nginx` — works only because that namespace is the documented exception.) ## Fix Select on `kubernetes.io/metadata.name: monitoring`. ## Impact Unblocks gateway metrics: `llm_ttft_seconds`, `llm_itl_seconds`, `llm_tokens_total`, `queue_full_requests_total`, `warmup_requests_delayed_total`, `vllm_prefix_cache_hit_tokens_total`. These are the only per-request LLM latency series that cover the two Ollama models (`ornith:35b`, `qwen2.5:3b-instruct`), which expose no metrics endpoint of their own. --------- Co-authored-by: Admin Bot <[email protected]> Reviewed-on: #62	2026-09-29 14:40:31 +00:00
rock	3068c68730	Merge pull request 'feat(sessions): SandboxFleet integration — fleet primary, k8s fallback' (#61 ) from feat/sessions-sandbox-fleet-integration into main CI / CI (push) Successful in 3m28s Details	2026-09-29 08:14:00 +00:00