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

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
 }
@@ -127,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.
@@ -218,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,
@@ -462,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,
@@ -567,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,
@@ -585,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.
@@ -768,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
 		}
@@ -786,52 +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, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
-			resp.Body.Close()
-			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"