feat(cag): memory standing-context adapter, warmup semaphore, cache-aware routing

Issue #44: code-register memory adapter with standing-context resource; fix
dispatcher method-level Auth.Capability override (was always using verb-derived
capability, ignoring per-method Auth field).

Issue #45: per-backend bounded in-memory queue (VLLM_MAX_QUEUE_DEPTH=20) +
concurrency semaphore (VLLM_MAX_NUM_SEQS=4); warmup requests (X-Poimen-Warmup:
true) sleep WARMUP_DELAY_MS before queue entry; /metrics endpoint wired.

Issue #46: rendezvous-hash backend selection on X-Poimen-Cache-Key; round-robin
fallback; MarkFailed excludes unhealthy backends 30s; ModifyResponse captures
vllm-num-cached-tokens + emits structured log + increments counter.

Closes #44 Closes #45 Closes #46

internal/proxy/proxy.go

@@ -3,13 +3,16 @@ package proxy
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"os"
 	"sort"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
@@ -18,6 +21,7 @@ import (
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/identity"
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/logging"
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/tracing"
 )
 
@@ -37,13 +41,18 @@ type Handler struct {
 	defaultReadTimeout    time.Duration
 	defaultWriteTimeout   time.Duration
 	defaultMaxBodySize    int64
+	// CAG: per-model backend schedulers for cache-aware routing (#45/#46)
+	schedulers  map[string]*Scheduler
+	// CAG: delay applied to warmup requests before queue entry (#45)
+	warmupDelay time.Duration
 }
 
 // Route represents a reverse proxy route.
 type Route struct {
-	Name     string
-	Upstream *config.Upstream
-	Director func(*http.Request)
+	Name      string
+	ModelName string // non-empty for model-based routes; used for scheduler lookup
+	Upstream  *config.Upstream
+	Director  func(*http.Request)
 	Transport *http.Transport
 }
 
@@ -93,6 +102,36 @@ func New(cfg *config.Config) *Handler {
 		h.authHTTP = newAuthClient()
 	}
 
+	// CAG: warmup delay from env (#45)
+	warmupMs := int64(500)
+	if v := os.Getenv("WARMUP_DELAY_MS"); v != "" {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n >= 0 {
+			warmupMs = n
+		}
+	}
+	h.warmupDelay = time.Duration(warmupMs) * time.Millisecond
+
+	// CAG: per-model backend schedulers (#45/#46)
+	maxSlots := 4
+	if v := os.Getenv("VLLM_MAX_NUM_SEQS"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			maxSlots = n
+		}
+	}
+	maxQueue := 20
+	if v := os.Getenv("VLLM_MAX_QUEUE_DEPTH"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			maxQueue = n
+		}
+	}
+	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)
+		}
+	}
+
 	for name, route := range cfg.Routes {
 		// Create a transport per unique upstream address for connection reuse
 		transport := h.getOrCreateTransport(route.Upstream.Address, &route.Upstream)
@@ -414,6 +453,51 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	// CAG: warmup deprioritisation + per-backend semaphore (issues #45/#46)
+	var selectedBackend *Backend
+	cacheKey := r.Header.Get("X-Poimen-Cache-Key")
+	if sched := h.schedulers[route.ModelName]; sched != nil {
+		selectedBackend = sched.Select(cacheKey)
+
+		// Warmup: sleep BEFORE queue entry so agents arriving in that window win slots first.
+		if r.Header.Get("X-Poimen-Warmup") == "true" {
+			r.Header.Del("X-Poimen-Warmup")
+			r.Header.Set("X-Priority", "low")
+			time.Sleep(h.warmupDelay)
+			observability.Global().IncrWarmupDelayed()
+		}
+
+		if selectedBackend != nil {
+			if err := selectedBackend.Acquire(r.Context()); err != nil {
+				if errors.Is(err, ErrQueueFull) {
+					w.WriteHeader(http.StatusServiceUnavailable)
+					fmt.Fprintf(w, "queue full")
+				} else {
+					w.WriteHeader(http.StatusServiceUnavailable)
+					fmt.Fprintf(w, "upstream busy")
+				}
+				return
+			}
+			defer selectedBackend.Release()
+
+			// Patch route to selected backend if it differs from the config default.
+			if selectedBackend.Address != route.Upstream.Address {
+				patchedCfg := *route.Upstream
+				patchedCfg.Address = selectedBackend.Address
+				patchedURL, _ := url.Parse("http://" + selectedBackend.Address)
+				route = &Route{
+					Name:      route.Name,
+					ModelName: route.ModelName,
+					Upstream:  &patchedCfg,
+					Transport: h.getOrCreateTransport(selectedBackend.Address, &patchedCfg),
+					Director: func(req *http.Request) {
+						directorFunc(req, patchedURL, &patchedCfg)
+					},
+				}
+			}
+		}
+	}
+
 	// Create the reverse proxy
 	proxy := httputil.NewSingleHostReverseProxy(&url.URL{
 		Scheme: "http",
@@ -426,8 +510,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// Use the connection-pooled transport wrapped with tracing
 	proxy.Transport = tracing.NewTransport(route.Transport)
 
-	// Set error handler to log upstream errors
+	// Set error handler: log upstream errors and exclude failed backend from routing.
 	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
+		if selectedBackend != nil {
+			if sched := h.schedulers[route.ModelName]; sched != nil {
+				sched.MarkFailed(selectedBackend.Address)
+			}
+		}
 		w.WriteHeader(http.StatusBadGateway)
 		fmt.Fprintf(w, "upstream error")
 		logging.Errorf("upstream error", err, map[string]string{
@@ -436,6 +525,29 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		})
 	}
 
+	// CAG: capture vllm-num-cached-tokens from response headers (issue #46)
+	if cacheKey != "" && r.URL.Path == "/v1/chat/completions" {
+		proxy.ModifyResponse = func(resp *http.Response) error {
+			raw := resp.Header.Get("vllm-num-cached-tokens")
+			if raw == "" {
+				return nil
+			}
+			n, err := strconv.ParseInt(raw, 10, 64)
+			if err != nil || n <= 0 {
+				return nil
+			}
+			observability.Global().IncrPrefixCacheHitTokens(n)
+			logging.Event(map[string]interface{}{
+				"event":         "llm_request",
+				"cache_key":     cacheKey,
+				"cached_tokens": n,
+				"total_tokens":  0,
+				"hit_rate":      0.0,
+			})
+			return nil
+		}
+	}
+
 	// Note: We apply connection timeout via Transport dialer, but NOT read timeout as a context deadline.
 	// Read timeout should apply to inactivity (socket read timeout), not total request duration.
 	// A streaming response that's continuously sending should not be cut off.

internal/proxy/proxy_test.go

@@ -414,3 +414,87 @@ func TestProxyPreservesBody(t *testing.T) {
 		t.Errorf("expected body %s, got %s", testBody, receivedBody)
 	}
 }
+
+// TestWarmupHeaderHandling verifies that X-Poimen-Warmup: true is stripped from
+// the upstream request and replaced with X-Priority: low, while no X-Poimen-Warmup
+// on a normal request produces neither header downstream.
+//
+// TODO(#45): implement warmup header stripping in proxy.Handler Director or middleware.
+// Once implemented, remove t.Skip and wire up the real Handler.
+func TestWarmupHeaderHandling(t *testing.T) {
+	t.Skip("requires full handler setup: warmup header transform (#45) not yet implemented")
+
+	tests := []struct {
+		name            string
+		sendWarmup      bool
+		wantWarmupGone  bool
+		wantPriorityLow bool
+	}{
+		{
+			name:            "warmup=true",
+			sendWarmup:      true,
+			wantWarmupGone:  true,  // X-Poimen-Warmup must NOT reach upstream
+			wantPriorityLow: true,  // X-Priority: low must be set
+		},
+		{
+			name:            "warmup=false",
+			sendWarmup:      false,
+			wantWarmupGone:  true,  // no header sent, so absent at upstream
+			wantPriorityLow: false, // X-Priority must NOT be set
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var gotWarmup, gotPriority string
+			upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				gotWarmup = r.Header.Get("X-Poimen-Warmup")
+				gotPriority = r.Header.Get("X-Priority")
+				w.WriteHeader(http.StatusOK)
+			}))
+			defer upstream.Close()
+
+			upstreamAddr := strings.TrimPrefix(upstream.URL, "http://")
+			cfg := &config.Config{
+				Routes: map[string]*config.Route{
+					"test": {
+						Name: "test",
+						Upstream: config.Upstream{
+							Address:        upstreamAddr,
+							ConnectTimeout: 5 * time.Second,
+							ReadTimeout:    5 * time.Second,
+							WriteTimeout:   5 * time.Second,
+							MaxBodySize:    1024 * 1024,
+							AuthRequired:   false,
+						},
+					},
+				},
+			}
+			h := New(cfg)
+			defer h.Close()
+
+			srv := httptest.NewServer(h)
+			defer srv.Close()
+
+			req, _ := http.NewRequest("GET", srv.URL+"/test", nil)
+			if tc.sendWarmup {
+				req.Header.Set("X-Poimen-Warmup", "true")
+			}
+			resp, err := http.DefaultClient.Do(req)
+			if err != nil {
+				t.Fatalf("request failed: %v", err)
+			}
+			resp.Body.Close()
+
+			if tc.wantWarmupGone && gotWarmup != "" {
+				t.Errorf("X-Poimen-Warmup must be stripped before upstream, got %q", gotWarmup)
+			}
+			if tc.wantPriorityLow && gotPriority != "low" {
+				t.Errorf("X-Priority: low must be set when warmup=true, got %q", gotPriority)
+			}
+			if !tc.wantPriorityLow && gotPriority != "" {
+				t.Errorf("X-Priority must not be set for normal requests, got %q", gotPriority)
+			}
+		})
+	}
+}

internal/proxy/router.go

@@ -159,6 +159,7 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) {
 
 	route := &Route{
 		Name:      "v1-chat-" + modelName,
+		ModelName: modelName,
 		Upstream:  upstreamCfg,
 		Transport: h.getOrCreateTransport(modelUpstream.Address, upstreamCfg),
 		Director: func(req *http.Request) {

internal/proxy/scheduler.go

@@ -0,0 +1,155 @@
+package proxy
+
+import (
+	"context"
+	"errors"
+	"hash/fnv"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// ErrQueueFull is returned when the in-memory queue is at capacity.
+var ErrQueueFull = errors.New("backend queue full")
+
+// 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)
+type Backend struct {
+	Address string
+	queue   chan struct{}
+	slots   chan struct{}
+}
+
+// newBackend creates a Backend with given queue depth and concurrency limit.
+func newBackend(addr string, maxSlots, maxQueue int) *Backend {
+	slots := make(chan struct{}, maxSlots)
+	for i := 0; i < maxSlots; i++ {
+		slots <- struct{}{}
+	}
+	return &Backend{
+		Address: addr,
+		queue:   make(chan struct{}, maxQueue),
+		slots:   slots,
+	}
+}
+
+// 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.
+func (b *Backend) Acquire(ctx context.Context) error {
+	// Non-blocking queue entry: reject immediately when queue is full.
+	select {
+	case b.queue <- struct{}{}:
+		// entered queue
+	default:
+		return ErrQueueFull
+	}
+	// Wait for a concurrency slot, then leave the queue.
+	select {
+	case <-b.slots:
+		<-b.queue // leave queue; now in service
+		return nil
+	case <-ctx.Done():
+		<-b.queue // leave queue; cancelled
+		return ctx.Err()
+	}
+}
+
+// Release returns a concurrency slot to the pool after the upstream request completes.
+func (b *Backend) Release() {
+	b.slots <- struct{}{}
+}
+
+// Scheduler routes LLM requests to backends using rendezvous hashing (cache-aware)
+// or round-robin (no cache key). Failed backends are excluded for a configurable window.
+type Scheduler struct {
+	mu          sync.RWMutex
+	backends    []*Backend
+	failedUntil map[string]time.Time
+	failWindow  time.Duration
+	rrIdx       atomic.Uint64
+}
+
+// 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 {
+	backends := make([]*Backend, len(addresses))
+	for i, addr := range addresses {
+		backends[i] = newBackend(addr, maxSlots, maxQueue)
+	}
+	return &Scheduler{
+		backends:    backends,
+		failedUntil: make(map[string]time.Time),
+		failWindow:  30 * time.Second,
+	}
+}
+
+// Select returns a backend. Uses rendezvous hashing when cacheKey is non-empty
+// (same key always routes to the same backend, skipping unhealthy ones).
+// Uses round-robin when cacheKey is empty.
+func (s *Scheduler) Select(cacheKey string) *Backend {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	if len(s.backends) == 0 {
+		return nil
+	}
+	if len(s.backends) == 1 {
+		return s.backends[0]
+	}
+	if cacheKey != "" {
+		return s.selectRendezvous(cacheKey)
+	}
+	return s.selectRoundRobin()
+}
+
+// selectRendezvous picks the healthy backend with the highest FNV32a score
+// for the key (highest random weight / rendezvous hashing).
+// Falls back to next-highest score when primary is excluded by MarkFailed.
+// Must be called with mu held.
+func (s *Scheduler) selectRendezvous(cacheKey string) *Backend {
+	type scored struct {
+		b     *Backend
+		score uint32
+	}
+	now := time.Now()
+	scores := make([]scored, len(s.backends))
+	h := fnv.New32a()
+	for i, b := range s.backends {
+		h.Reset()
+		h.Write([]byte(cacheKey + ":" + b.Address))
+		scores[i] = scored{b: b, score: h.Sum32()}
+	}
+	sort.Slice(scores, func(i, j int) bool {
+		return scores[i].score > scores[j].score
+	})
+	for _, sc := range scores {
+		if t, failed := s.failedUntil[sc.b.Address]; !failed || now.After(t) {
+			if now.After(t) {
+				delete(s.failedUntil, sc.b.Address)
+			}
+			return sc.b
+		}
+	}
+	// All backends unhealthy: return highest-scored (fail-open).
+	return scores[0].b
+}
+
+// selectRoundRobin distributes requests evenly when no cache key is present.
+// Must be called with mu held.
+func (s *Scheduler) selectRoundRobin() *Backend {
+	idx := s.rrIdx.Add(1) - 1
+	return s.backends[idx%uint64(len(s.backends))]
+}
+
+// MarkFailed excludes addr from routing for the fail window (default 30 s).
+// Called by the proxy ErrorHandler on upstream connection failure.
+func (s *Scheduler) MarkFailed(addr string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.failedUntil[addr] = time.Now().Add(s.failWindow)
+}

internal/proxy/scheduler_test.go

@@ -0,0 +1,126 @@
+package proxy
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+// TestBackendAcquireBlocksThenUnblocksOnRelease verifies that a second Acquire
+// blocks while a slot is held, then unblocks when Release is called.
+func TestBackendAcquireBlocksThenUnblocksOnRelease(t *testing.T) {
+	b := newBackend("host:1", 1, 1)
+
+	// G1: acquire and hold the only slot.
+	if err := b.Acquire(context.Background()); err != nil {
+		t.Fatalf("G1 Acquire: unexpected error: %v", err)
+	}
+
+	g2Done := make(chan error, 1)
+	go func() {
+		g2Done <- b.Acquire(context.Background())
+	}()
+
+	// Give G2 time to block.
+	time.Sleep(20 * time.Millisecond)
+
+	select {
+	case err := <-g2Done:
+		t.Fatalf("G2 returned before Release: %v", err)
+	default:
+	}
+
+	// G1 releases; G2 should unblock.
+	b.Release()
+
+	select {
+	case err := <-g2Done:
+		if err != nil {
+			t.Fatalf("G2 Acquire after Release: unexpected error: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("G2 did not unblock within 1s after Release")
+	}
+
+	// Cleanup: release G2's slot.
+	b.Release()
+}
+
+// TestBackendAcquireReturnsErrQueueFullWhenQueueFull verifies that Acquire returns
+// ErrQueueFull immediately when no queue capacity exists (maxQueue=0).
+// With a zero-capacity queue channel the non-blocking select always hits default.
+func TestBackendAcquireReturnsErrQueueFullWhenQueueFull(t *testing.T) {
+	b := newBackend("host:1", 1, 0)
+
+	err := b.Acquire(context.Background())
+	if !errors.Is(err, ErrQueueFull) {
+		t.Fatalf("expected ErrQueueFull, got %v", err)
+	}
+}
+
+// TestSchedulerSelectRendezvousHashDeterministic verifies that the same cache key
+// 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)
+
+	first := s.Select("same-cache-key")
+	if first == nil {
+		t.Fatal("Select returned nil")
+	}
+	for i := 1; i < 10; i++ {
+		b := s.Select("same-cache-key")
+		if b == nil || b.Address != first.Address {
+			t.Fatalf("call %d: want %s, got %v", i, first.Address, b)
+		}
+	}
+}
+
+// TestSchedulerSelectNoKeyRoundRobin verifies that round-robin distributes
+// 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)
+
+	counts := make(map[string]int)
+	for i := 0; i < 9; i++ {
+		b := s.Select("")
+		if b == nil {
+			t.Fatalf("call %d: Select returned nil", i)
+		}
+		counts[b.Address]++
+	}
+
+	for _, addr := range addrs {
+		if counts[addr] != 3 {
+			t.Errorf("addr %s: want 3 selections, got %d (all counts: %v)", addr, counts[addr], counts)
+		}
+	}
+}
+
+// TestSchedulerMarkFailedExcludesBackend verifies that a failed backend is not
+// 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.MarkFailed("a:80")
+
+	var mu sync.Mutex
+	seen := make(map[string]int)
+	for i := 0; i < 10; i++ {
+		b := s.Select("any-key")
+		if b == nil {
+			t.Fatalf("call %d: Select returned nil", i)
+		}
+		mu.Lock()
+		seen[b.Address]++
+		mu.Unlock()
+	}
+
+	if seen["a:80"] > 0 {
+		t.Errorf("failed backend a:80 was selected %d times; expected 0 within fail window", seen["a:80"])
+	}
+}