fix(proxy): admission-control review fixes missing from the #65 merge (#66)

Follow-up to #65, which is already merged. The review of that PR returned BLOCK with two high-severity defects; the fix commit was pushed **after** the merge landed, so none of it is in `main`.

## What happened

- `main` @ `369d51a` is the squash of the first commit only (`c5a91cf`).
- `d71c884` (review fixes) is on the old branch and was never merged — PR #65 is closed.
- Verified against `main` directly: `ADMISSION_MAX_INFLIGHT` 0 hits, `func (rw *responseWriter) Flush` 0 hits, `ModifyResponse` 0 hits, `Snapshot` 0 hits.
- The image in production (`forgejo.riotpiao.com/rock/api-gateway@sha256:64f2bfc1…`, 3/3 replicas) is built from that squash, so the defects below are **live right now**.

## Live state confirmed

`https://api.riotpiao.com/metrics` on the running pods:

```
observer_degraded 0
vllm_requests_running 1
vllm_kv_usage_perc 0.008293092588704587
admission_parked_total 0
engine_state_stale_total 0
```

The observer itself works: it is reading the real vLLM exposition body and the startup self-check passed (`observer_degraded 0`). No requests have parked yet, so the two defects below have not been observable in the wild — they are proven from code, not from a live failure.

## Defects fixed here

**1. The parked-stream keep-alive never runs in production (high).** `internal/tracing/middleware.go` wraps the `ResponseWriter` in a struct that embeds the interface and implements only `WriteHeader`/`Write`, so `w.(http.Flusher)` in `internal/proxy/proxy.go` is false for every real request (`cmd/gateway/main.go` → `tracing.Middleware` → `router`). `prewritten` therefore never becomes true, the heartbeat never starts, and a parked streaming request writes zero bytes for up to `ADMISSION_MAX_PARK_SECONDS` (120) — the client timeout the feature exists to prevent. Fixed by adding `Flush()` and `Unwrap()` to the wrapper.

**2. Upstream status masked once the SSE status is spent (high, latent behind 1).** `httputil.ReverseProxy` calls `rw.WriteHeader(res.StatusCode)` unconditionally; after the park wrote 200, net/http keeps 200 and the upstream error body is appended raw into the event stream, which a streaming client reads as an empty completion and reports as success. Fixed with a `ModifyResponse` that frames an upstream `>= 400` as an `event: error` when the status has already been written.

**3. Lost wakeup in the park loop.** `Get()` and `Tick()` were taken under two separate lock acquisitions, so a poll landing between them pushed the waiter onto the next generation's channel — up to one extra poll interval (2s) parked. `Snapshot()` now returns state and channel under one lock.

**4. `observer_degraded` pinned at 1 forever.** It was raised only on the *first* poll's parse failure and `SetObserverDegraded(false)` had no call site, so one truncated response pinned the gauge for the pod's life, while genuine metric-name drift appearing later was never reported at all. `markDegraded` now tracks current parsability in both directions and only touches the exporter on a change.

**5. `ADMISSION_MAX_INFLIGHT` silently defaulted.** Now set explicitly to `4` in `k8s/deployment.yaml` with the coupling to the engine's `--max-num-seqs` documented, instead of quietly inheriting `VLLM_MAX_NUM_SEQS`.

## Verification

`go build ./...` clean · `go vet ./...` clean · `go test ./...` 368 passed, 0 failed · `structural-check.sh` 0 errors (1 pre-existing `:latest` warning) · `go.mod` unchanged.

Cherry-picked onto `main` rather than merged from the old branch, so this diff is exactly the five fixes and nothing else: 5 files, +80/−8.

## Still open

The reviewer's BLOCK also rested on nine functions over CRAP>30 purely from 0% coverage — the deliberate "no new tests" decision. Worth recording that defect 1 is a one-line type assertion that a `httptest` case through the real middleware chain would have caught immediately. Raising it here so the decision is visible rather than implied.

---------

Co-authored-by: Poimen <[email protected]>
Reviewed-on: #66
Co-authored-by: poimen <[email protected]>

cmd/gateway/main.go

@@ -4,9 +4,13 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"math"
 	"net/http"
 	"os"
 	"os/signal"
+	"runtime"
+	"strconv"
+	"strings"
 	"syscall"
 
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
@@ -21,6 +25,10 @@ import (
 )
 
 func main() {
+	// Size the Go scheduler to the container's CPU quota before any server
+	// goroutine starts.
+	configureGOMAXPROCS()
+
 	ctx := context.Background()
 
 	// Initialize OpenTelemetry tracing
@@ -177,3 +185,53 @@ func main() {
 	log.Printf("gateway shutdown complete")
 	os.Exit(0)
 }
+
+// configureGOMAXPROCS sets GOMAXPROCS from the cgroup CPU quota so the runtime
+// sizes its scheduler to the container limit rather than the host CPU count.
+// The default is left alone when no quota is configured.
+func configureGOMAXPROCS() {
+	quota, period := readCPUQuota()
+	if quota <= 0 || period <= 0 {
+		return
+	}
+	n := int(math.Ceil(float64(quota) / float64(period)))
+	if n < 1 {
+		n = 1
+	}
+	runtime.GOMAXPROCS(n)
+	log.Printf("GOMAXPROCS set to %d (cgroup cpu quota %d/%d)", n, quota, period)
+}
+
+// readCPUQuota returns the CPU quota and period from cgroup v2, falling back to
+// cgroup v1. A quota of 0 means no limit.
+func readCPUQuota() (int64, int64) {
+	// cgroup v2: "<quota> <period>" or "max <period>".
+	if data, err := os.ReadFile("/sys/fs/cgroup/cpu.max"); err == nil {
+		fields := strings.Fields(string(data))
+		if len(fields) >= 2 {
+			period, perr := strconv.ParseInt(fields[1], 10, 64)
+			if perr == nil && period > 0 {
+				if fields[0] == "max" {
+					return 0, period
+				}
+				if quota, qerr := strconv.ParseInt(fields[0], 10, 64); qerr == nil {
+					return quota, period
+				}
+			}
+		}
+	}
+
+	// cgroup v1: cpu.cfs_quota_us is -1 when unlimited.
+	period := int64(100000)
+	if data, err := os.ReadFile("/sys/fs/cgroup/cpu/cpu.cfs_period_us"); err == nil {
+		if v, perr := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64); perr == nil && v > 0 {
+			period = v
+		}
+	}
+	if data, err := os.ReadFile("/sys/fs/cgroup/cpu/cpu.cfs_quota_us"); err == nil {
+		if v, qerr := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64); qerr == nil {
+			return v, period
+		}
+	}
+	return 0, 0
+}

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,22 +41,33 @@ type Metrics struct {
 	warmupRequestsDelayed int64
 	prefixCacheHitTokens  int64
 	queueFullRequests     int64
+
+	// Admission control metrics (#64)
+	admissionParked   int64
+	admissionRejected map[string]int64
+	engineStateStale  int64
+
+	// Engine gauge readings mirrored from vLLM (#64)
+	vllmRequestsRunning int64
+	vllmKVUsagePerc     float64
+	observerDegraded    int
 }
 
 // NewMetrics creates a new Metrics instance.
 func NewMetrics() *Metrics {
 	return &Metrics{
-		requestTotal:           make(map[string]int64),
-		requestDuration:        make(map[string]int64),
-		requestDurationBuckets: make(map[string]map[float64]int64),
-		bytesIn:                make(map[string]int64),
-		bytesOut:               make(map[string]int64),
-		upstreamHealth:         make(map[string]int),
+		requestTotal:            make(map[string]int64),
+		requestDuration:         make(map[string]int64),
+		requestDurationBuckets:  make(map[string]map[float64]int64),
+		bytesIn:                 make(map[string]int64),
+		bytesOut:                make(map[string]int64),
+		upstreamHealth:          make(map[string]int),
 		streamingResponsesTotal: make(map[string]int64),
-		streamingByteCount:     make(map[string]int64),
-		ttftMs:                 make(map[string][]int64),
-		itlMs:                  make(map[string][]int64),
-		tokenCount:             make(map[string]int64),
+		streamingByteCount:      make(map[string]int64),
+		ttftMs:                  make(map[string][]int64),
+		itlMs:                   make(map[string][]int64),
+		tokenCount:              make(map[string]int64),
+		admissionRejected:       make(map[string]int64),
 	}
 }
 
@@ -288,6 +299,12 @@ func (m *Metrics) Reset() {
 	m.warmupRequestsDelayed = 0
 	m.prefixCacheHitTokens = 0
 	m.queueFullRequests = 0
+	m.admissionParked = 0
+	m.admissionRejected = make(map[string]int64)
+	m.engineStateStale = 0
+	m.vllmRequestsRunning = 0
+	m.vllmKVUsagePerc = 0
+	m.observerDegraded = 0
 }
 
 // IncrQueueFull increments queue_full_requests_total.
@@ -332,6 +349,98 @@ func (m *Metrics) GetPrefixCacheHitTokens() int64 {
 	return m.prefixCacheHitTokens
 }
 
+// IncrAdmissionParked increments admission_parked_total.
+func (m *Metrics) IncrAdmissionParked() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.admissionParked++
+}
+
+// GetAdmissionParked returns admission_parked_total.
+func (m *Metrics) GetAdmissionParked() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.admissionParked
+}
+
+// IncrAdmissionRejected increments admission_rejected_total{reason=reason}.
+func (m *Metrics) IncrAdmissionRejected(reason string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.admissionRejected[reason]++
+}
+
+// GetAdmissionRejected returns a copy of admission_rejected_total by reason.
+func (m *Metrics) GetAdmissionRejected() map[string]int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	out := make(map[string]int64, len(m.admissionRejected))
+	for reason, count := range m.admissionRejected {
+		out[reason] = count
+	}
+	return out
+}
+
+// IncrEngineStateStale increments engine_state_stale_total.
+func (m *Metrics) IncrEngineStateStale() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.engineStateStale++
+}
+
+// GetEngineStateStale returns engine_state_stale_total.
+func (m *Metrics) GetEngineStateStale() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.engineStateStale
+}
+
+// SetVLLMRequestsRunning sets vllm_requests_running from the gateway's own reading.
+func (m *Metrics) SetVLLMRequestsRunning(n int64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.vllmRequestsRunning = n
+}
+
+// GetVLLMRequestsRunning returns vllm_requests_running.
+func (m *Metrics) GetVLLMRequestsRunning() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.vllmRequestsRunning
+}
+
+// SetVLLMKVUsagePerc sets vllm_kv_usage_perc from the gateway's own reading.
+func (m *Metrics) SetVLLMKVUsagePerc(v float64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.vllmKVUsagePerc = v
+}
+
+// GetVLLMKVUsagePerc returns vllm_kv_usage_perc.
+func (m *Metrics) GetVLLMKVUsagePerc() float64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.vllmKVUsagePerc
+}
+
+// SetObserverDegraded sets observer_degraded (1 = startup self-check failed).
+func (m *Metrics) SetObserverDegraded(degraded bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if degraded {
+		m.observerDegraded = 1
+	} else {
+		m.observerDegraded = 0
+	}
+}
+
+// GetObserverDegraded returns observer_degraded.
+func (m *Metrics) GetObserverDegraded() int {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.observerDegraded
+}
+
 // globalMetrics is the singleton Metrics instance for gateway-wide telemetry.
 var (
 	globalMetrics     *Metrics

internal/observability/prometheus.go

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

internal/proxy/engine_state.go

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

internal/proxy/proxy.go

@@ -2,6 +2,7 @@
 package proxy
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -47,6 +48,10 @@ type Handler struct {
 	schedulers  map[string]*Scheduler
 	// CAG: delay applied to warmup requests before queue entry (#45)
 	warmupDelay time.Duration
+	// engineObservers polls each model backend's /metrics for admission (#64)
+	engineObservers map[string]*EngineObserver
+	// maxBodyBytes caps synthesized model-route request bodies (MAX_BODY_BYTES)
+	maxBodyBytes int64
 	// transportsMu guards h.transports for goroutine-safe reads/writes.
 	transportsMu sync.RWMutex
 }
@@ -122,18 +127,80 @@ func New(cfg *config.Config) *Handler {
 			maxSlots = n
 		}
 	}
-	maxQueue := 20
+	maxQueue := 8
 	if v := os.Getenv("VLLM_MAX_QUEUE_DEPTH"); v != "" {
 		if n, err := strconv.Atoi(v); err == nil && n > 0 {
 			maxQueue = n
 		}
 	}
+
+	// Issue #64: engine-wide admission thresholds. MaxInFlight defaults to the
+	// same value as VLLM_MAX_NUM_SEQS; the observers below read the engine's
+	// own gauges, which every replica shares.
+	admissionMaxInFlight := maxSlots
+	if v := os.Getenv("ADMISSION_MAX_INFLIGHT"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			admissionMaxInFlight = n
+		}
+	}
+	admissionMaxKVUsage := 0.90
+	if v := os.Getenv("ADMISSION_MAX_KV_USAGE"); v != "" {
+		if f, err := strconv.ParseFloat(v, 64); err == nil && f > 0 {
+			admissionMaxKVUsage = f
+		}
+	}
+	maxParkSeconds := 120
+	if v := os.Getenv("ADMISSION_MAX_PARK_SECONDS"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			maxParkSeconds = n
+		}
+	}
+	h.maxBodyBytes = 1048576
+	if v := os.Getenv("MAX_BODY_BYTES"); v != "" {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 {
+			h.maxBodyBytes = n
+		}
+	}
+
+	limits := AdmissionLimits{
+		MaxInFlight: admissionMaxInFlight,
+		MaxKVUsage:  admissionMaxKVUsage,
+		MaxPark:     time.Duration(maxParkSeconds) * time.Second,
+	}
+
+	// Issue #64: only backends that actually publish the vLLM scheduler gauges
+	// may be observed. `reasoning` is the only vLLM engine here; ornith and
+	// qwen-cpu are Ollama and embeddings/rerank are TEI, none of which export
+	// vllm:* series. Observing them would trip the startup self-check and hold
+	// observer_degraded at 1 forever. Unset (the default) observes nothing and
+	// leaves admission exactly as it was before this change.
+	observeModels := make(map[string]bool)
+	for _, name := range strings.Split(os.Getenv("ADMISSION_OBSERVE_MODELS"), ",") {
+		if name = strings.TrimSpace(name); name != "" {
+			observeModels[name] = true
+		}
+	}
+
+	h.engineObservers = make(map[string]*EngineObserver)
 	h.schedulers = make(map[string]*Scheduler)
 	for modelName, model := range cfg.Models {
 		addrs := model.AllAddresses()
-		if len(addrs) > 0 {
-			h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue)
+		if len(addrs) == 0 {
+			continue
 		}
+		if observeModels[modelName] {
+			for _, addr := range addrs {
+				if _, ok := h.engineObservers[addr]; !ok {
+					h.engineObservers[addr] = NewEngineObserver(addr, enginePollInterval)
+				}
+			}
+		}
+		h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue, h.engineObservers, limits)
+	}
+	// The handler has no request-path lifecycle to hang cancellation on, so the
+	// observers poll for the lifetime of the process.
+	for _, o := range h.engineObservers {
+		o.Start(context.Background())
 	}
 	// Pre-populate transports for all scheduler backends at startup (single-threaded).
 	// getOrCreateTransport is not goroutine-safe; pre-populating avoids concurrent writes.
@@ -151,7 +218,7 @@ func New(cfg *config.Config) *Handler {
 		// Create a transport per unique upstream address for connection reuse
 		transport := h.getOrCreateTransport(route.Upstream.Address, &route.Upstream)
 		
-		upstreamURL, _ := url.Parse("http://" + route.Upstream.Address)
+		upstreamURL := &url.URL{Scheme: "http", Host: route.Upstream.Address}
 		
 		r := &Route{
 			Name:      name,
@@ -395,27 +462,104 @@ func (h *Handler) applyWarmup(ctx context.Context, r *http.Request) bool {
 // acquireScheduledBackend selects a backend from sched, applies warmup
 // deprioritisation, acquires a concurrency slot, and patches route when the
 // selected backend differs from the config default. It returns the backend
-// to Release (caller must defer), the (possibly patched) route, and ok=true.
-// If the slot cannot be acquired it writes a 503 and returns (nil, nil, false).
-func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool) {
+// to Release (caller must defer), the (possibly patched) route, whether SSE
+// headers were already written while parking, and ok=true.
+// If the slot cannot be acquired it writes a 503 and returns ok=false.
+func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool, bool) {
 	backend := sched.Select(cacheKey)
 	if !h.applyWarmup(ctx, r) {
-		return nil, nil, false
+		return nil, nil, false, false
 	}
 	if backend == nil {
-		return nil, route, true
+		return nil, route, false, true
 	}
-	if err := backend.Acquire(ctx); err != nil {
-		if errors.Is(err, ErrQueueFull) {
+
+	// Decision Q22: a request that actually parks on the engine gate and wants
+	// an SSE stream gets its headers pre-written plus a 15 s keep-alive, so no
+	// intermediary times the connection out before the engine admits it.
+	var (
+		prewritten bool
+		hbStop     chan struct{}
+		hbDone     chan struct{}
+	)
+	onPark := func() {
+		if !requestWantsStream(r) {
+			return
+		}
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			return
+		}
+		prewritten = true
+		hbStop = make(chan struct{})
+		hbDone = make(chan struct{})
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("Connection", "keep-alive")
+		w.Header().Set("X-Accel-Buffering", "no")
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprint(w, ": keep-alive\n\n")
+		flusher.Flush()
+		go func() {
+			defer close(hbDone)
+			ticker := time.NewTicker(15 * time.Second)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-hbStop:
+					return
+				case <-ctx.Done():
+					return
+				case <-ticker.C:
+					fmt.Fprint(w, ": keep-alive\n\n")
+					flusher.Flush()
+				}
+			}
+		}()
+	}
+
+	err := backend.acquire(ctx, onPark)
+	if prewritten {
+		close(hbStop)
+		<-hbDone
+	}
+	if err != nil {
+		// Once the SSE status line is spent, report the rejection as an event
+		// instead of a second WriteHeader.
+		reject := func(body string) {
+			if prewritten {
+				fmt.Fprintf(w, "event: error\ndata: {\"error\":%q}\n\n", body)
+				if f, ok := w.(http.Flusher); ok {
+					f.Flush()
+				}
+				return
+			}
+			w.WriteHeader(http.StatusServiceUnavailable)
+			fmt.Fprintf(w, "%s", body)
+		}
+		if errors.Is(err, ErrAdmissionTimeout) {
+			observability.Global().IncrAdmissionRejected("park_timeout")
+			logging.Warnf("admission timeout", map[string]string{
+				"backend": backend.Address,
+				"path":    r.URL.Path,
+				"model":   route.ModelName,
+			})
+			if !prewritten {
+				w.Header().Set("Retry-After", "5")
+			}
+			reject("admission timeout")
+		} else if errors.Is(err, ErrQueueFull) {
 			observability.Global().IncrQueueFull()
+			observability.Global().IncrAdmissionRejected("queue_full")
 			logging.Warnf("backend queue full", map[string]string{
 				"backend": backend.Address,
 				"path":    r.URL.Path,
 				"model":   route.ModelName,
 			})
-			w.Header().Set("Retry-After", "5")
-			w.WriteHeader(http.StatusServiceUnavailable)
-			fmt.Fprintf(w, "queue full")
+			if !prewritten {
+				w.Header().Set("Retry-After", "5")
+			}
+			reject("queue full")
 		} else {
 			logging.Warnf("backend acquire failed", map[string]string{
 				"backend": backend.Address,
@@ -423,15 +567,14 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr
 				"model":   route.ModelName,
 				"err":     err.Error(),
 			})
-			w.WriteHeader(http.StatusServiceUnavailable)
-			fmt.Fprintf(w, "upstream busy")
+			reject("upstream busy")
 		}
-		return nil, nil, false
+		return nil, nil, prewritten, false
 	}
 	if backend.Address != route.Upstream.Address {
 		patchedCfg := *route.Upstream
 		patchedCfg.Address = backend.Address
-		patchedURL, _ := url.Parse("http://" + backend.Address)
+		patchedURL := &url.URL{Scheme: "http", Host: backend.Address}
 		route = &Route{
 			Name:      route.Name,
 			ModelName: route.ModelName,
@@ -442,7 +585,27 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr
 			},
 		}
 	}
-	return backend, route, true
+	return backend, route, prewritten, true
+}
+
+// requestWantsStream reports whether the JSON request body asks for an SSE
+// stream ("stream": true). The body is restored for the upstream.
+func requestWantsStream(r *http.Request) bool {
+	if r.Body == nil {
+		return false
+	}
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		return false
+	}
+	r.Body = io.NopCloser(bytes.NewReader(body))
+	var probe struct {
+		Stream bool `json:"stream"`
+	}
+	if err := json.Unmarshal(body, &probe); err != nil {
+		return false
+	}
+	return probe.Stream
 }
 
 // handleAuthEndpoint routes /auth/* requests and returns true if handled.
@@ -605,9 +768,10 @@ func (h *Handler) enforceBodyLimit(w http.ResponseWriter, r *http.Request, route
 // Returns (nil, nil, false) when the backend slot cannot be acquired (response written).
 func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *http.Request, route *Route, sched *Scheduler, cacheKey string) (*httputil.ReverseProxy, *Backend, bool) {
 	var selectedBackend *Backend
+	var headersPrewritten bool
 	if sched != nil {
 		var ok bool
-		selectedBackend, route, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey)
+		selectedBackend, route, headersPrewritten, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey)
 		if !ok {
 			return nil, nil, false
 		}
@@ -622,12 +786,58 @@ func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *h
 
 	capturedBackend := selectedBackend
 	capturedRoute := route
+	if headersPrewritten {
+		// The 200 + text/event-stream status is already on the wire, so an
+		// upstream status can no longer be forwarded: httputil.ReverseProxy
+		// calls WriteHeader again and net/http keeps the first one. Frame a
+		// failure as an SSE event rather than letting a raw error body land
+		// inside the stream, which a streaming client reads as an empty
+		// completion and reports as success.
+		proxy.ModifyResponse = func(resp *http.Response) error {
+			if resp.StatusCode < http.StatusBadRequest {
+				return nil
+			}
+			body, readErr := io.ReadAll(io.LimitReader(resp.Body, 4096))
+			resp.Body.Close()
+			if readErr != nil {
+				logging.Warnf("reading upstream error body failed", map[string]string{
+					"upstream": capturedRoute.Upstream.Address,
+					"err":      readErr.Error(),
+				})
+			}
+			logging.Warnf("upstream error after parked stream started", map[string]string{
+				"upstream": capturedRoute.Upstream.Address,
+				"status":   strconv.Itoa(resp.StatusCode),
+			})
+			resp.Body = io.NopCloser(strings.NewReader(fmt.Sprintf(
+				"event: error\ndata: {\"upstream_status\":%d,\"error\":%q}\n\n",
+				resp.StatusCode, strings.TrimSpace(string(body)))))
+			resp.ContentLength = -1
+			resp.Header.Del("Content-Length")
+			resp.Header.Set("Content-Type", "text/event-stream")
+			return nil
+		}
+	}
 	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
 		if capturedBackend != nil {
 			if s := h.schedulers[capturedRoute.ModelName]; s != nil {
 				s.MarkFailed(capturedBackend.Address)
 			}
 		}
+		if headersPrewritten {
+			// SSE headers and a 200 are already on the wire: a 502 here would be
+			// an ignored WriteHeader and would tear the event stream. Report the
+			// failure as an event instead.
+			fmt.Fprint(w, "event: error\ndata: {\"error\":\"upstream error\"}\n\n")
+			if f, ok := w.(http.Flusher); ok {
+				f.Flush()
+			}
+			logging.Errorf("upstream error", err, map[string]string{
+				"upstream": capturedRoute.Upstream.Address,
+				"path":     r.URL.Path,
+			})
+			return
+		}
 		w.WriteHeader(http.StatusBadGateway)
 		fmt.Fprintf(w, "upstream error")
 		logging.Errorf("upstream error", err, map[string]string{

internal/proxy/router.go

@@ -69,9 +69,10 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	// Restore the body so it can be read again by the upstream
 	r.Body = io.NopCloser(bytes.NewReader(bodyBytes))
 
-	// Parse the JSON to find the model field
-	var payload map[string]interface{}
-	if err := json.Unmarshal(bodyBytes, &payload); err != nil {
+	// Extract only the top-level "model" value with a token walk: the common
+	// path must not materialise the whole body (messages and all) as a map.
+	modelRaw, hasModel, err := extractTopLevelModel(bodyBytes)
+	if err != nil {
 		return nil, &modelValidationError{
 			Kind:    "invalid_json",
 			Message: "request body is not valid JSON",
@@ -79,7 +80,6 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	}
 
 	// Extract the model name
-	modelVal, hasModel := payload["model"]
 	if !hasModel {
 		return nil, &modelValidationError{
 			Kind:    "missing_model",
@@ -88,7 +88,7 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	}
 
 	// Handle null model
-	if modelVal == nil {
+	if bytes.Equal(bytes.TrimSpace(modelRaw), []byte("null")) {
 		return nil, &modelValidationError{
 			Kind:    "null_model",
 			Message: "'model' field is null",
@@ -96,8 +96,8 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 	}
 
 	// Extract as string
-	modelName, ok := modelVal.(string)
-	if !ok {
+	var modelName string
+	if err := json.Unmarshal(modelRaw, &modelName); err != nil {
 		return nil, &modelValidationError{
 			Kind:    "missing_model",
 			Message: "'model' field must be a string",
@@ -122,10 +122,26 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 		}
 	}
 
-	// If upstream expects a different model name, rewrite the body
+	// If upstream expects a different model name, rewrite the body. This path is
+	// uncommon, so it may decode the body as raw fields; the common path above
+	// does not build a full map.
 	if modelUpstream.UpstreamModel != "" && modelUpstream.UpstreamModel != modelName {
-		payload["model"] = modelUpstream.UpstreamModel
-		newBody, err := json.Marshal(payload)
+		var fields map[string]json.RawMessage
+		if err := json.Unmarshal(bodyBytes, &fields); err != nil {
+			return nil, &modelValidationError{
+				Kind:    "invalid_request",
+				Message: fmt.Sprintf("failed to rewrite model name: %v", err),
+			}
+		}
+		newModel, err := json.Marshal(modelUpstream.UpstreamModel)
+		if err != nil {
+			return nil, &modelValidationError{
+				Kind:    "invalid_request",
+				Message: fmt.Sprintf("failed to rewrite model name: %v", err),
+			}
+		}
+		fields["model"] = newModel
+		newBody, err := json.Marshal(fields)
 		if err != nil {
 			return nil, &modelValidationError{
 				Kind:    "invalid_request",
@@ -151,11 +167,11 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 		ConnectTimeout: h.defaultConnectTimeout,
 		ReadTimeout:    h.defaultReadTimeout,
 		WriteTimeout:   h.defaultWriteTimeout,
-		MaxBodySize:    h.defaultMaxBodySize,
+		MaxBodySize:    h.maxBodyBytes,
 		AuthRequired:   false,
 	}
 
-	targetURL, _ := url.Parse("http://" + modelUpstream.Address)
+	targetURL := &url.URL{Scheme: "http", Host: modelUpstream.Address}
 
 	route := &Route{
 		Name:      "v1-chat-" + modelName,
@@ -169,3 +185,87 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 
 	return route, nil
 }
+
+// extractTopLevelModel walks a JSON object body with a streaming decoder and
+// returns the raw value of the top-level "model" field. It never builds the
+// full document as a map, so a large "messages" array is skipped by token.
+// hasModel is false when the key is absent or the document is a bare null.
+func extractTopLevelModel(body []byte) (model json.RawMessage, hasModel bool, err error) {
+	dec := json.NewDecoder(bytes.NewReader(body))
+	tok, err := dec.Token()
+	if err != nil {
+		return nil, false, err
+	}
+	if tok == nil {
+		// A bare `null` body decoded to a nil map before, i.e. no model field.
+		if _, err := dec.Token(); err != io.EOF {
+			if err == nil {
+				return nil, false, fmt.Errorf("unexpected data after JSON null")
+			}
+			return nil, false, err
+		}
+		return nil, false, nil
+	}
+	if delim, ok := tok.(json.Delim); !ok || delim != '{' {
+		return nil, false, fmt.Errorf("request body is not a JSON object")
+	}
+	for dec.More() {
+		keyTok, err := dec.Token()
+		if err != nil {
+			return nil, false, err
+		}
+		key, _ := keyTok.(string)
+		if key == "model" {
+			if err := dec.Decode(&model); err != nil {
+				return nil, false, err
+			}
+			hasModel = true
+			continue
+		}
+		if err := skipJSONValue(dec); err != nil {
+			return nil, false, err
+		}
+	}
+	// Consume the closing brace and reject trailing data, preserving the
+	// invalid_json answer the old whole-body decode produced.
+	if _, err := dec.Token(); err != nil {
+		return nil, false, err
+	}
+	if _, err := dec.Token(); err != io.EOF {
+		if err == nil {
+			return nil, false, fmt.Errorf("unexpected data after JSON object")
+		}
+		return nil, false, err
+	}
+	return model, hasModel, nil
+}
+
+// skipJSONValue advances dec past the next JSON value without materialising it.
+func skipJSONValue(dec *json.Decoder) error {
+	tok, err := dec.Token()
+	if err != nil {
+		return err
+	}
+	delim, ok := tok.(json.Delim)
+	if !ok {
+		return nil // scalar
+	}
+	if delim != '{' && delim != '[' {
+		return fmt.Errorf("unexpected delimiter %v", delim)
+	}
+	depth := 1
+	for depth > 0 {
+		tok, err := dec.Token()
+		if err != nil {
+			return err
+		}
+		if d, ok := tok.(json.Delim); ok {
+			if d == '{' || d == '[' {
+				depth++
+			} else {
+				depth--
+			}
+		}
+	}
+	return nil
+}

internal/proxy/scheduler.go

@@ -8,20 +8,41 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
 )
 
 // ErrQueueFull is returned when the in-memory queue is at capacity.
 var ErrQueueFull = errors.New("backend queue full")
 
+// ErrAdmissionTimeout is returned when a request parked on engine admission
+// exceeds the configured maximum park duration.
+var ErrAdmissionTimeout = errors.New("admission timeout")
+
+// AdmissionLimits are the engine-wide admission thresholds applied to every
+// backend of a scheduler (#64).
+type AdmissionLimits struct {
+	MaxInFlight int
+	MaxKVUsage  float64
+	MaxPark     time.Duration
+}
+
 // Backend represents a single vLLM upstream with a bounded in-memory queue.
 // Two channels model the queue:
 //
 //	queue  — waiting requests (bounded by VLLM_MAX_QUEUE_DEPTH)
 //	slots  — concurrent in-service requests (bounded by VLLM_MAX_NUM_SEQS)
+//
+// The engine gate is applied before the queue: admission reads the engine's
+// own published gauges via observer, which is global across replicas.
 type Backend struct {
-	Address string
-	queue   chan struct{}
-	slots   chan struct{}
+	Address     string
+	queue       chan struct{}
+	slots       chan struct{}
+	observer    *EngineObserver
+	maxInFlight int
+	maxKVUsage  float64
+	maxPark     time.Duration
 }
 
 // newBackend creates a Backend with given queue depth and concurrency limit.
@@ -40,7 +61,48 @@ func newBackend(addr string, maxSlots, maxQueue int) *Backend {
 // Acquire enters the in-memory queue then waits for a vLLM concurrency slot.
 // Returns ErrQueueFull immediately if the queue channel is at capacity (no goroutine queued).
 // Returns ctx.Err() if the context is cancelled while waiting for a slot.
+// Returns ErrAdmissionTimeout if the engine gate parks the request past maxPark.
 func (b *Backend) Acquire(ctx context.Context) error {
+	return b.acquire(ctx, nil)
+}
+
+// acquire is Acquire with an optional onPark callback. onPark is invoked once,
+// on the calling goroutine, the first time the request has to wait for engine
+// admission; the parked streaming path uses it to start an SSE heartbeat.
+func (b *Backend) acquire(ctx context.Context, onPark func()) error {
+	// Engine gate (#64): admit on the engine's own gauges, not on this pod's
+	// counters. Failing OPEN on !Healthy is deliberate — a metrics outage must
+	// not become an inference outage; the three-replica gate is only a fairness
+	// improvement on top of the engine's own queueing.
+	if b.observer != nil {
+		deadline := time.Now().Add(b.maxPark)
+		timer := time.NewTimer(time.Until(deadline))
+		defer timer.Stop()
+		parked := false
+		for {
+			// State and tick channel are read together under one lock: taking
+			// them separately can miss a poll that lands between the two calls.
+			st, tick := b.observer.Snapshot()
+			if !st.Healthy || (st.InFlight() < int64(b.maxInFlight) && st.KVUsage < b.maxKVUsage) {
+				break
+			}
+			if !parked {
+				observability.Global().IncrAdmissionParked()
+				parked = true
+				if onPark != nil {
+					onPark()
+				}
+			}
+			select {
+			case <-tick:
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-timer.C:
+				return ErrAdmissionTimeout
+			}
+		}
+	}
+
 	// Non-blocking queue entry: reject immediately when queue is full.
 	select {
 	case b.queue <- struct{}{}:
@@ -76,10 +138,15 @@ type Scheduler struct {
 
 // NewScheduler creates a Scheduler for the given addresses.
 // maxSlots = VLLM_MAX_NUM_SEQS, maxQueue = VLLM_MAX_QUEUE_DEPTH.
-func NewScheduler(addresses []string, maxSlots, maxQueue int) *Scheduler {
+// observers holds one engine observer per backend address (may be nil).
+func NewScheduler(addresses []string, maxSlots, maxQueue int, observers map[string]*EngineObserver, limits AdmissionLimits) *Scheduler {
 	backends := make([]*Backend, len(addresses))
 	for i, addr := range addresses {
 		backends[i] = newBackend(addr, maxSlots, maxQueue)
+		backends[i].observer = observers[addr]
+		backends[i].maxInFlight = limits.MaxInFlight
+		backends[i].maxKVUsage = limits.MaxKVUsage
+		backends[i].maxPark = limits.MaxPark
 	}
 	return &Scheduler{
 		backends:    backends,

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)
+	s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{})
 
 	first := s.Select("same-cache-key")
 	if first == nil {
@@ -85,7 +85,7 @@ func TestSchedulerSelectRendezvousHashDeterministic(t *testing.T) {
 // requests evenly across all backends when no cache key is given.
 func TestSchedulerSelectNoKeyRoundRobin(t *testing.T) {
 	addrs := []string{"a:80", "b:80", "c:80"}
-	s := NewScheduler(addrs, 10, 10)
+	s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{})
 
 	counts := make(map[string]int)
 	for i := 0; i < 9; i++ {
@@ -160,7 +160,7 @@ func TestBackendAcquireCancelledContext(t *testing.T) {
 // selected within the fail window (30 s default).
 func TestSchedulerMarkFailedExcludesBackend(t *testing.T) {
 	addrs := []string{"a:80", "b:80"}
-	s := NewScheduler(addrs, 10, 10)
+	s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{})
 
 	s.MarkFailed("a:80")
 

internal/tracing/middleware.go

@@ -36,6 +36,21 @@ func (rw *responseWriter) Write(b []byte) (int, error) {
 	return n, err
 }
 
+// Flush forwards to the wrapped writer so that streaming handlers can still
+// detect http.Flusher through this wrapper. Without it a bare
+// `w.(http.Flusher)` assertion fails for every request that passes through
+// this middleware, which silently disables SSE flushing and keep-alives.
+func (rw *responseWriter) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+// Unwrap exposes the wrapped writer to http.ResponseController.
+func (rw *responseWriter) Unwrap() http.ResponseWriter {
+	return rw.ResponseWriter
+}
+
 // Middleware returns an HTTP middleware that adds tracing to requests.
 func Middleware(next http.Handler) http.Handler {
 	tracer := otel.Tracer(tracerName)

k8s/deployment.yaml

@@ -21,9 +21,10 @@ spec:
       labels:
         app: api-gateway
         component: gateway
-        # Required by the llm-serving-default-deny NetworkPolicy, which admits
-        # only pods labelled llm-client=true (from any namespace) on port 8080.
-        # Without it every upstream dial times out and dispatch returns 502.
+        # The llm-serving-ingress NetworkPolicy in the llm-serving namespace now
+        # selects by namespace (kubernetes.io/metadata.name: api), not by this
+        # label. Kept for older NetworkPolicy revisions that still admit on
+        # llm-client=true.
         llm-client: "true"
       annotations:
         reloader.stakater.com/auto: "true"
@@ -67,6 +68,24 @@ spec:
           value: "5m"
         - name: LOG_LEVEL
           value: "info"
+        - name: ADMISSION_MAX_PARK_SECONDS
+          value: "120"
+        # Must track the engine's --max-num-seqs (homelab: reasoning.yaml).
+        # The gauge this compares against is engine-wide, so this number is a
+        # ceiling on concurrent requests across all gateway replicas, not a
+        # per-pod slot count. Drift here is silent; startup logs the value.
+        - name: ADMISSION_MAX_INFLIGHT
+          value: "4"
+        # Only backends that publish the vLLM scheduler gauges may be observed.
+        # `reasoning` is the only vLLM engine; ornith/qwen-cpu are Ollama and
+        # embeddings/rerank are TEI, so observing them would hold
+        # observer_degraded at 1 forever.
+        - name: ADMISSION_OBSERVE_MODELS
+          value: "reasoning"
+        - name: VLLM_MAX_QUEUE_DEPTH
+          value: "8"
+        - name: MAX_BODY_BYTES
+          value: "1048576"
         # OpenTelemetry tracing configuration
         - name: OTEL_EXPORTER_OTLP_ENDPOINT
           value: "otel-collector.tracing.svc.cluster.local:4317"