fix(proxy): wire LLM latency metrics into the streaming response path

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. 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.

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.
- 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 (issue #33) is unaffected.

Tests: unit coverage for SSE vs non-streaming, wrapper transparency, and an
end-to-end handler test that fails if the wrapper is ever wired nowhere again.

cmd/gateway/main.go

@@ -4,13 +4,9 @@ import (
 	"context"
 	"fmt"
 	"log"
-	"math"
 	"net/http"
 	"os"
 	"os/signal"
-	"runtime"
-	"strconv"
-	"strings"
 	"syscall"
 
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
@@ -25,10 +21,6 @@ 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
@@ -185,53 +177,3 @@ 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
-}

go.mod

@@ -5,51 +5,39 @@ 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/gorilla/securecookie v1.1.2 // indirect
+	github.com/google/uuid v1.6.0 // indirect
-	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // 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/oauth2 v0.36.0 // indirect
+	golang.org/x/net v0.58.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

go.sum

@@ -1,19 +1,11 @@
 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=
@@ -31,12 +23,8 @@ 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=
@@ -45,10 +33,6 @@ 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=
@@ -57,8 +41,6 @@ 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=
@@ -66,12 +48,6 @@ 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=
@@ -111,8 +87,6 @@ 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=

internal/auth/jwe_test.go

@@ -1,21 +0,0 @@
-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)
-	}
-}

internal/auth/jwt.go

@@ -3,141 +3,37 @@ 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"
 )
 
-// issuerMatches reports whether iss is the configured issuer or a sub-path of
+// isValidIssuer checks if issuer is from Authentik (any provider/app).
-// it. Zitadel emits exactly "https://zitadel.riotpiao.com"; providers that
+// Accepts: https://authentik.riotpiao.com/application/o/{provider}/
-// mount an application path append a suffix (Authentik used
+func isValidIssuer(iss string) bool {
-// "/application/o/<app>/"). A trailing slash on either side is ignored.
+	return strings.Contains(iss, "authentik.riotpiao.com/application/o/") &&
-//
+		strings.HasSuffix(iss, "/")
-// 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+"/")
 }
 
-// audienceValues normalises the aud claim. Zitadel access tokens carry an
+// Validator validates JWTs against Authentik JWKS.
-// array, older providers carry a single string.
+// Supports multi-issuer: any Authentik service account provider is accepted
-func audienceValues(claims jwt.MapClaims) ([]string, bool) {
+// (portfolio-agent, memory-agent, api-gw, etc.) because all share the same
-	switch v := claims["aud"].(type) {
+// JWKS signing key.
-	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 // Expected issuer (iss claim); empty disables the domain check
+	issuer   string // Not used for validation (kept for logging); issuer regex check is sufficient
-	audience string // Expected audience; currently logged only (any non-empty aud is accepted)
+	audience string // Not used for validation; any audience from valid Authentik issuer 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 is the expected iss claim (exact value or prefix). audience is kept
+// issuer and audience params are deprecated (ignored for validation) but kept
-// for compatibility; only non-emptiness is enforced. JWKS fetching is lazy
+// for backward compatibility. Multi-issuer validation via isValidIssuer() is used instead.
-// (deferred until first validation).
+// JWKS fetching is lazy (deferred until first validation).
 func NewValidator(issuer, audience, jwksURL string) *Validator {
 	return &Validator{
 		issuer:   issuer,   // deprecated param, kept for compat
@@ -191,31 +87,8 @@ func (v *Validator) ValidateBearerToken(authHeader string) (jwt.MapClaims, error
 	}
 
 	// Ensure JWKS is loaded (lazy)
-	if v.jwks != nil || v.zitadel == nil {
+	if err := v.ensureJWKS(); err != nil {
-		if err := v.ensureJWKS(); err != nil {
+		return nil, err
-			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
@@ -247,35 +120,34 @@ func (v *Validator) ValidateBearerToken(authHeader string) (jwt.MapClaims, error
 		}
 	}
 
-	// Check iss against the configured issuer (exact match or sub-path).
+	// 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
 	if iss, ok := claims["iss"].(string); !ok {
 		return nil, fmt.Errorf("missing issuer claim")
-	} else if !issuerMatches(iss, v.issuer) {
+	} else if !isValidIssuer(iss) {
-		return nil, fmt.Errorf("invalid issuer: %s (want %s)", iss, v.issuer)
+		return nil, fmt.Errorf("invalid issuer: %s", iss)
 	}
 
-	// Check aud - Zitadel sends an array; accept a string or array.
+	// Check aud (audience) - accept any Authentik-provided audience
-	if auds, ok := audienceValues(claims); !ok {
+	// 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 {
 		return nil, fmt.Errorf("missing audience claim")
-	} else if len(auds) == 0 {
+	} else if aud == "" {
 		return nil, fmt.Errorf("empty audience claim")
 	}
-	// The signature check plus the issuer check establish the trust boundary;
+	// Note: Not hardcoding expected audience. Any audience from a valid Authentik
-	// the audience value itself is not hardcoded.
+	// issuer is accepted, since all service accounts are under the same trust boundary.
 
 	return claims, nil
 }
 
 // CheckPermissions checks if claims contain required permission(s).
-// Understands three token shapes:
+// Checks both "permissions" claim (for users) and "roles" claim (for service accounts).
-//   - "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 {
-	// 1. permissions claim (array of strings)
+	// Try permissions claim first (for user tokens)
 	if permsIface, ok := claims["permissions"]; ok {
 		if perms, ok := permsIface.([]interface{}); ok {
 			if v.checkPermList(perms, required...) {
@@ -284,45 +156,15 @@ func (v *Validator) CheckPermissions(claims jwt.MapClaims, required ...string) b
 		}
 	}
 
-	// 2. roles claim: array (legacy) or map keyed by role name
+	// Fall back to roles claim (for service account tokens)
 	if rolesIface, ok := claims["roles"]; ok {
-		switch roles := rolesIface.(type) {
+		if roles, ok := rolesIface.([]interface{}); ok {
-		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
 }
 

internal/auth/zitadel.go

@@ -1,157 +0,0 @@
-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
-}

internal/auth/zitadel_test.go

@@ -1,294 +0,0 @@
-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()
-}

internal/config/config.go

@@ -66,23 +66,12 @@ type AuthConfig struct {
 	JWKSURL string
 	// RequiredCapability is the permission required for LLM inference (e.g., "llm:inference").
 	RequiredCapability string
-	// TokenURL is the OIDC token endpoint used for token exchange
+	// TokenURL is the Authentik token endpoint for password/refresh grants.
-	// (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.

internal/config/loader.go

@@ -20,16 +20,13 @@ type rawConfig struct {
 
 // rawAuth represents auth configuration in YAML.
 type rawAuth struct {
-	Enabled             bool     `yaml:"enabled"`
+	Enabled            bool   `yaml:"enabled"`
-	Issuer              string   `yaml:"issuer"`
+	Issuer             string `yaml:"issuer"`
-	Audience            string   `yaml:"audience"`
+	Audience           string `yaml:"audience"`
-	JWKSURL             string   `yaml:"jwksUrl"`
+	JWKSURL            string `yaml:"jwksUrl"`
-	RequiredCapability  string   `yaml:"requiredCapability"`
+	RequiredCapability string `yaml:"requiredCapability"`
-	TokenURL            string   `yaml:"tokenUrl"`
+	TokenURL           string `yaml:"tokenUrl"`
-	ClientID            string   `yaml:"clientId"`
+	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.
@@ -187,17 +184,14 @@ func LoadRoutesAndModelsFromFile(path string) (map[string]*Route, map[string]*Mo
 
 	// Parse auth config
 	authConfig := AuthConfig{
-		Enabled:             raw.Auth.Enabled,
+		Enabled:            raw.Auth.Enabled,
-		Issuer:              raw.Auth.Issuer,
+		Issuer:             raw.Auth.Issuer,
-		Audience:            raw.Auth.Audience,
+		Audience:           raw.Auth.Audience,
-		JWKSURL:             raw.Auth.JWKSURL,
+		JWKSURL:            raw.Auth.JWKSURL,
-		RequiredCapability:  raw.Auth.RequiredCapability,
+		RequiredCapability: raw.Auth.RequiredCapability,
-		TokenURL:            raw.Auth.TokenURL,
+		TokenURL:           raw.Auth.TokenURL,
-		ClientID:            raw.Auth.ClientID,
+		ClientID:           raw.Auth.ClientID,
-		ClientSecret:        os.Getenv("AUTH_CLIENT_SECRET"),
+		ClientSecret:       os.Getenv("AUTH_CLIENT_SECRET"),
-		ZitadelDomain:       raw.Auth.ZitadelDomain,
-		ZitadelInsecurePort: raw.Auth.ZitadelInsecurePort,
-		ZitadelClientIDs:    raw.Auth.ZitadelClientIDs,
 	}
 
 	return routes, models, adapters, authConfig, nil

internal/observability/metrics.go

@@ -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,33 +41,22 @@ 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),
+		requestTotal:           make(map[string]int64),
-		requestDuration:         make(map[string]int64),
+		requestDuration:        make(map[string]int64),
-		requestDurationBuckets:  make(map[string]map[float64]int64),
+		requestDurationBuckets: make(map[string]map[float64]int64),
-		bytesIn:                 make(map[string]int64),
+		bytesIn:                make(map[string]int64),
-		bytesOut:                make(map[string]int64),
+		bytesOut:               make(map[string]int64),
-		upstreamHealth:          make(map[string]int),
+		upstreamHealth:         make(map[string]int),
 		streamingResponsesTotal: make(map[string]int64),
-		streamingByteCount:      make(map[string]int64),
+		streamingByteCount:     make(map[string]int64),
-		ttftMs:                  make(map[string][]int64),
+		ttftMs:                 make(map[string][]int64),
-		itlMs:                   make(map[string][]int64),
+		itlMs:                  make(map[string][]int64),
-		tokenCount:              make(map[string]int64),
+		tokenCount:             make(map[string]int64),
-		admissionRejected:       make(map[string]int64),
 	}
 }
 
@@ -144,7 +133,7 @@ func (m *Metrics) RecordStreamingResponse(route, upstream string, totalBytes int
 
 func (m *Metrics) recordDuration(key string, duration time.Duration) {
 	m.requestDuration[key] += int64(duration.Milliseconds())
-
+	
 	if _, ok := m.requestDurationBuckets[key]; !ok {
 		m.requestDurationBuckets[key] = make(map[float64]int64)
 	}
@@ -299,12 +288,6 @@ 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.
@@ -349,98 +332,6 @@ 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

internal/observability/prometheus.go

@@ -48,38 +48,6 @@ 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"
 }
 

internal/proxy/engine_state.go

@@ -1,333 +0,0 @@
-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
-}

internal/proxy/proxy.go

@@ -2,7 +2,6 @@
 package proxy
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -48,10 +47,6 @@ 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
 }
@@ -98,29 +93,8 @@ func New(cfg *config.Config) *Handler {
 		defaultMaxBodySize:    100 * 1024 * 1024,
 	}
 
-	// Initialize JWT validator if auth is enabled.
+	// Initialize JWT validator if auth is enabled
-	//
+	if cfg.Auth.Enabled && cfg.Auth.JWKSURL != "" {
-	// 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,
@@ -148,80 +122,18 @@ func New(cfg *config.Config) *Handler {
 			maxSlots = n
 		}
 	}
-	maxQueue := 8
+	maxQueue := 20
 	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 {
+		if len(addrs) > 0 {
-			continue
+			h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue)
 		}
-		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.
@@ -239,7 +151,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.URL{Scheme: "http", Host: route.Upstream.Address}
+		upstreamURL, _ := url.Parse("http://" + route.Upstream.Address)
 		
 		r := &Route{
 			Name:      name,
@@ -483,104 +395,27 @@ 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, whether SSE
+// to Release (caller must defer), the (possibly patched) route, and ok=true.
-// headers were already written while parking, and ok=true.
+// If the slot cannot be acquired it writes a 503 and returns (nil, nil, false).
-// 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) {
-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, false
+		return nil, nil, false
 	}
 	if backend == nil {
-		return nil, route, false, true
+		return nil, route, true
 	}
-
+	if err := backend.Acquire(ctx); err != nil {
-	// Decision Q22: a request that actually parks on the engine gate and wants
+		if errors.Is(err, ErrQueueFull) {
-	// 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,
 			})
-			if !prewritten {
+			w.Header().Set("Retry-After", "5")
-				w.Header().Set("Retry-After", "5")
+			w.WriteHeader(http.StatusServiceUnavailable)
-			}
+			fmt.Fprintf(w, "queue full")
-			reject("queue full")
 		} else {
 			logging.Warnf("backend acquire failed", map[string]string{
 				"backend": backend.Address,
@@ -588,14 +423,15 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr
 				"model":   route.ModelName,
 				"err":     err.Error(),
 			})
-			reject("upstream busy")
+			w.WriteHeader(http.StatusServiceUnavailable)
+			fmt.Fprintf(w, "upstream busy")
 		}
-		return nil, nil, prewritten, false
+		return nil, nil, false
 	}
 	if backend.Address != route.Upstream.Address {
 		patchedCfg := *route.Upstream
 		patchedCfg.Address = backend.Address
-		patchedURL := &url.URL{Scheme: "http", Host: backend.Address}
+		patchedURL, _ := url.Parse("http://" + backend.Address)
 		route = &Route{
 			Name:      route.Name,
 			ModelName: route.ModelName,
@@ -606,27 +442,7 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr
 			},
 		}
 	}
-	return backend, route, prewritten, true
+	return backend, route, 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.
@@ -743,17 +559,16 @@ func (h *Handler) authenticateV1(w http.ResponseWriter, r *http.Request) bool {
 		return false
 	}
 	identity.Inject(r, claims)
-	if caps := requiredCapabilities(h.config.Auth.RequiredCapability); len(caps) > 0 {
+	if h.config.Auth.RequiredCapability != "" {
-		args := append(caps, "*")
+		if !h.jwtValidator.CheckPermissions(claims, h.config.Auth.RequiredCapability, "*") {
-		if !h.jwtValidator.CheckPermissions(claims, args...) {
 			writeProblemDetail(w, http.StatusForbidden,
 				"https://api.example.com/problems/insufficient-permissions",
 				"Insufficient Permissions",
-				fmt.Sprintf("Required capability: %s", strings.Join(caps, " or ")),
+				fmt.Sprintf("Required capability: %s", h.config.Auth.RequiredCapability),
 				nil)
 			logging.Errorf("auth failed", fmt.Errorf("insufficient permissions"), map[string]string{
 				"path":     r.URL.Path,
-				"required": strings.Join(caps, ","),
+				"required": h.config.Auth.RequiredCapability,
 			})
 			return false
 		}
@@ -761,22 +576,6 @@ 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 {
@@ -806,10 +605,9 @@ 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, headersPrewritten, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey)
+		selectedBackend, route, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey)
 		if !ok {
 			return nil, nil, false
 		}
@@ -824,58 +622,12 @@ 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{

internal/proxy/router.go

@@ -69,10 +69,9 @@ 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))
 
-	// Extract only the top-level "model" value with a token walk: the common
+	// Parse the JSON to find the model field
-	// path must not materialise the whole body (messages and all) as a map.
+	var payload map[string]interface{}
-	modelRaw, hasModel, err := extractTopLevelModel(bodyBytes)
+	if err := json.Unmarshal(bodyBytes, &payload); err != nil {
-	if err != nil {
 		return nil, &modelValidationError{
 			Kind:    "invalid_json",
 			Message: "request body is not valid JSON",
@@ -80,6 +79,7 @@ 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 bytes.Equal(bytes.TrimSpace(modelRaw), []byte("null")) {
+	if modelVal == nil {
 		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
-	var modelName string
+	modelName, ok := modelVal.(string)
-	if err := json.Unmarshal(modelRaw, &modelName); err != nil {
+	if !ok {
 		return nil, &modelValidationError{
 			Kind:    "missing_model",
 			Message: "'model' field must be a string",
@@ -122,26 +122,10 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 		}
 	}
 
-	// If upstream expects a different model name, rewrite the body. This path is
+	// If upstream expects a different model name, rewrite the body
-	// 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 {
-		var fields map[string]json.RawMessage
+		payload["model"] = modelUpstream.UpstreamModel
-		if err := json.Unmarshal(bodyBytes, &fields); err != nil {
+		newBody, err := json.Marshal(payload)
-			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",
@@ -167,11 +151,11 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 		ConnectTimeout: h.defaultConnectTimeout,
 		ReadTimeout:    h.defaultReadTimeout,
 		WriteTimeout:   h.defaultWriteTimeout,
-		MaxBodySize:    h.maxBodyBytes,
+		MaxBodySize:    h.defaultMaxBodySize,
 		AuthRequired:   false,
 	}
 
-	targetURL := &url.URL{Scheme: "http", Host: modelUpstream.Address}
+	targetURL, _ := url.Parse("http://" + modelUpstream.Address)
 
 	route := &Route{
 		Name:      "v1-chat-" + modelName,
@@ -185,87 +169,3 @@ 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
-}

internal/proxy/scheduler.go

@@ -8,41 +8,20 @@ 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
+	Address string
-	queue       chan struct{}
+	queue   chan struct{}
-	slots       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.
@@ -61,48 +40,7 @@ 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{}{}:
@@ -138,15 +76,10 @@ type Scheduler struct {
 
 // NewScheduler creates a Scheduler for the given addresses.
 // maxSlots = VLLM_MAX_NUM_SEQS, maxQueue = VLLM_MAX_QUEUE_DEPTH.
-// observers holds one engine observer per backend address (may be nil).
+func NewScheduler(addresses []string, maxSlots, maxQueue int) *Scheduler {
-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,

internal/proxy/scheduler_test.go

@@ -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, nil, AdmissionLimits{})
+	s := NewScheduler(addrs, 10, 10)
 
 	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, nil, AdmissionLimits{})
+	s := NewScheduler(addrs, 10, 10)
 
 	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, nil, AdmissionLimits{})
+	s := NewScheduler(addrs, 10, 10)
 
 	s.MarkFailed("a:80")
 

internal/tracing/middleware.go

@@ -36,21 +36,6 @@ 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)

k8s/deployment.yaml

@@ -21,10 +21,9 @@ spec:
       labels:
         app: api-gateway
         component: gateway
-        # The llm-serving-ingress NetworkPolicy in the llm-serving namespace now
+        # Required by the llm-serving-default-deny NetworkPolicy, which admits
-        # selects by namespace (kubernetes.io/metadata.name: api), not by this
+        # only pods labelled llm-client=true (from any namespace) on port 8080.
-        # label. Kept for older NetworkPolicy revisions that still admit on
+        # Without it every upstream dial times out and dispatch returns 502.
-        # llm-client=true.
         llm-client: "true"
       annotations:
         reloader.stakater.com/auto: "true"
@@ -68,24 +67,6 @@ 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"

k8s/gateway-config-secret.yaml

@@ -7,4 +7,4 @@ metadata:
         app: api-gateway
 type: Opaque
 stringData:
-    config.yaml: "# PRODUCTION GATEWAY CONFIGURATION\n# ==========================================\n# All upstream services MUST use Kubernetes internal service DNS names\n# Format: <service>.<namespace>.svc.cluster.local\n# \n# This ensures:\n#  - Communication within cluster network only (no external IP exposure)\n#  - Pod-to-pod service discovery via internal DNS\n#  - Security policy enforcement at network level\n#  - Service-level load balancing via kube-proxy\n#\n# Routing Pattern:\n#  PREFERRED: X-Service header routing (e.g., X-Service: workflow)\n#  Legacy: Path-based routing (e.g., /workflow) - being deprecated\n#\nauth:\n  enabled: true\n  # ZITADEL is the only identity provider. CoreDNS rewrites\n  # zitadel.riotpiao.com to ingress-nginx, so discovery and JWKS stay inside\n  # the cluster and TLS is the wildcard certificate.\n  issuer: \"https://zitadel.riotpiao.com\"\n  # zitadelDomain enables the official ZITADEL Go SDK path: OIDC discovery,\n  # remote JWKS cache, and op.AccessTokenVerifier.\n  zitadelDomain: \"zitadel.riotpiao.com\"\n  # Audience allow-list: a caller token must carry one of these client IDs.\n  # Add a consumer here before it may call the LLM models.\n  zitadelClientIds:\n  - \"396757800874672511\"   # api-gw\n  - \"396757803559027071\"   # paperless (paperless-ai-agent)\n  - \"396757803877794175\"   # poimen\n  - \"396757801210282367\"   # argocd\n  - \"396757804213403649\"   # queue-sqs\n  # Two spellings accepted: ZITADEL role keys cannot carry a colon,\n  # Authentik permission strings could.\n  requiredCapability: \"llm-inference,llm:inference\"\n  # token exchange (client_credentials) against the same instance\n  tokenUrl: \"https://zitadel.riotpiao.com/oauth/v2/token\"\n  clientId: \"396757800874672511\"\n  # jwksUrl is intentionally unset: the SDK discovers it.\nroutes: []\nmodels:\n# All model services use internal Kubernetes DNS (llm-serving namespace)\n- name: \"reasoning\"\n  address: \"reasoning-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/chat/completions\"\n- name: \"ornith:35b\"\n  address: \"ornith-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/chat/completions\"\n- name: \"qwen2.5:3b-instruct\"\n  address: \"qwen-cpu.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/chat/completions\"\n- name: \"nomic-ai/nomic-embed-text-v2-moe\"\n  address: \"embeddings-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/embeddings\"\n- name: \"BAAI/bge-reranker-base\"\n  address: \"reranker-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/rerank\"\nadapters:\n- serviceName: sqs\n  upstream:\n    url: http://management-service.sqs.svc.cluster.local:9090\n    timeoutSeconds: 30\n  auth:\n    required: true\n  resources:\n  - name: send-message\n    methods:\n    - verb: POST\n      upstreamPath: /sqs/send\n  - name: receive-message\n    methods:\n    - verb: POST\n      upstreamPath: /sqs/receive\n  - name: list-queues\n    methods:\n    - verb: GET\n      upstreamPath: /sqs/queues\n- serviceName: workflow\n  upstream:\n    url: grpc://temporal-frontend.temporal.svc.cluster.local:7233\n    timeoutSeconds: 60\n  auth:\n    required: false\n  resources:\n  - name: execute\n    methods:\n    - verb: POST\n      upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ExecuteWorkflow\n  - name: describe\n    methods:\n    - verb: GET\n      upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/DescribeWorkflowExecution\n  - name: list\n    methods:\n    - verb: GET\n      upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ListWorkflowExecutions\n- serviceName: memory\n  upstream:\n    url: http://poimen-memory.poimen.svc.cluster.local:8080\n    timeoutSeconds: 30\n  auth:\n    required: false\n  resources:\n  - name: query\n    methods:\n    - verb: POST\n      upstreamPath: /memory/query\n  - name: ingest\n    methods:\n    - verb: POST\n      upstreamPath: /memory/ingest\n  - name: skills\n    methods:\n    - verb: GET\n      upstreamPath: /memory/skills\n- serviceName: s3\n  upstream:\n    url: http://minio.storage.svc.cluster.local:80\n    timeoutSeconds: 30\n  auth:\n    required: false\n  resources:\n  - name: list-objects\n    methods:\n    - verb: GET\n      upstreamPath: /\n  - name: get-object\n    methods:\n    - verb: GET\n      upstreamPath: /\n  - name: put-object\n    methods:\n    - verb: PUT\n      upstreamPath: /\n- serviceName: iam\n  upstream:\n    url: http://zitadel.iam.svc.cluster.local:8080\n    timeoutSeconds: 30\n  auth:\n    required: false\n  resources:\n  - name: list-roles\n    methods:\n    - verb: GET\n      upstreamPath: /api/v3/roles\n  - name: list-users\n    methods:\n    - verb: GET\n      upstreamPath: /api/v3/users\n  - name: create-role\n    methods:\n    - verb: POST\n      upstreamPath: /api/v3/roles\n"
+    config.yaml: "# PRODUCTION GATEWAY CONFIGURATION\n# ==========================================\n# All upstream services MUST use Kubernetes internal service DNS names\n# Format: <service>.<namespace>.svc.cluster.local\n# \n# This ensures:\n#  - Communication within cluster network only (no external IP exposure)\n#  - Pod-to-pod service discovery via internal DNS\n#  - Security policy enforcement at network level\n#  - Service-level load balancing via kube-proxy\n#\n# Routing Pattern:\n#  PREFERRED: X-Service header routing (e.g., X-Service: workflow)\n#  Legacy: Path-based routing (e.g., /workflow) - being deprecated\n#\nauth:\n  enabled: true\n  issuer: \"https://authentik.riotpiao.com/application/o/api-gw/\"\n  audience: \"api-gw\"\n  jwksUrl: \"http://authentik-server.iam.svc.cluster.local/application/o/api-gw/jwks/\"\n  requiredCapability: \"llm:inference\"\n  tokenUrl: \"http://authentik-server.iam.svc.cluster.local/application/o/token/\"\n  clientId: \"api-gw\"\nroutes: []\nmodels:\n# All model services use internal Kubernetes DNS (llm-serving namespace)\n- name: \"reasoning\"\n  address: \"reasoning-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/chat/completions\"\n- name: \"ornith:35b\"\n  address: \"ornith-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/chat/completions\"\n- name: \"qwen2.5:3b-instruct\"\n  address: \"qwen-cpu.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/chat/completions\"\n- name: \"nomic-ai/nomic-embed-text-v2-moe\"\n  address: \"embeddings-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/embeddings\"\n- name: \"BAAI/bge-reranker-base\"\n  address: \"reranker-predictor.llm-serving.svc.cluster.local:80\"\n  path: \"/v1/rerank\"\nadapters:\n- serviceName: sqs\n  upstream:\n    url: http://management-service.sqs.svc.cluster.local:9090\n    timeoutSeconds: 30\n  auth:\n    required: true\n  resources:\n  - name: send-message\n    methods:\n    - verb: POST\n      upstreamPath: /sqs/send\n  - name: receive-message\n    methods:\n    - verb: POST\n      upstreamPath: /sqs/receive\n  - name: list-queues\n    methods:\n    - verb: GET\n      upstreamPath: /sqs/queues\n- serviceName: workflow\n  upstream:\n    url: grpc://temporal-frontend.temporal.svc.cluster.local:7233\n    timeoutSeconds: 60\n  auth:\n    required: false\n  resources:\n  - name: execute\n    methods:\n    - verb: POST\n      upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ExecuteWorkflow\n  - name: describe\n    methods:\n    - verb: GET\n      upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/DescribeWorkflowExecution\n  - name: list\n    methods:\n    - verb: GET\n      upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ListWorkflowExecutions\n- serviceName: memory\n  upstream:\n    url: http://poimen-memory.poimen.svc.cluster.local:8080\n    timeoutSeconds: 30\n  auth:\n    required: false\n  resources:\n  - name: query\n    methods:\n    - verb: POST\n      upstreamPath: /memory/query\n  - name: ingest\n    methods:\n    - verb: POST\n      upstreamPath: /memory/ingest\n  - name: skills\n    methods:\n    - verb: GET\n      upstreamPath: /memory/skills\n- serviceName: s3\n  upstream:\n    url: http://minio.storage.svc.cluster.local:80\n    timeoutSeconds: 30\n  auth:\n    required: false\n  resources:\n  - name: list-objects\n    methods:\n    - verb: GET\n      upstreamPath: /\n  - name: get-object\n    methods:\n    - verb: GET\n      upstreamPath: /\n  - name: put-object\n    methods:\n    - verb: PUT\n      upstreamPath: /\n- serviceName: iam\n  upstream:\n    url: http://authentik-server.iam.svc.cluster.local:80\n    timeoutSeconds: 30\n  auth:\n    required: false\n  resources:\n  - name: list-roles\n    methods:\n    - verb: GET\n      upstreamPath: /api/v3/roles\n  - name: list-users\n    methods:\n    - verb: GET\n      upstreamPath: /api/v3/users\n  - name: create-role\n    methods:\n    - verb: POST\n      upstreamPath: /api/v3/roles\n"

k8s/network-policy.yaml

@@ -99,21 +99,19 @@ spec:
     ports:
     - protocol: TCP
       port: 8080
-  # Allow to IAM (Zitadel) for JWKS fetch + API
+  # Allow to IAM (Authentik) for JWKS fetch + API
-  # Zitadel API listens on 8080 (h2c/plaintext in-cluster); the login UI
+  # Authentik pod listens on 9000 (http) and 9443 (https)
-  # listens on 3000. Public access goes through ingress-nginx on 443 (already
+  # Service translates 80→9000, 443→9443
-  # covered by the 10.96.0.0/12 CIDR below), but 8080 keeps direct in-cluster
+  # NetworkPolicy matches destination pod port, not service port
-  # 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: 8080
+      port: 9000
     - protocol: TCP
-      port: 3000
+      port: 9443
   # Allow to SQS (queue management service)
   - to:
     - namespaceSelector: