feat(routing): add sticky runtime backends

2026-05-29 07:43:29 +08:00
parent 9d92360401
commit 98bd619ec8
11 changed files with 2046 additions and 5 deletions
--- a/internal/routing/sticky.go
+++ b/internal/routing/sticky.go
@@ -0,0 +1,199 @@
+package routing
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+)
+
+const (
+	RuntimeBackendMemory = "memory"
+	RuntimeBackendRedis  = "redis"
+
+	StickyScopeConversation = "conversation"
+	StickyScopeSession      = "session"
+	StickyScopeUser         = "user"
+)
+
+type StickyBinding struct {
+	LogicalGroupID string `json:"logical_group_id"`
+	PublicModel    string `json:"public_model"`
+	RouteID        string `json:"route_id"`
+	ShadowGroupID  string `json:"shadow_group_id"`
+	BoundAt        string `json:"bound_at,omitempty"`
+	ExpiresAt      string `json:"expires_at,omitempty"`
+}
+
+type RouteFailureState struct {
+	RouteID        string `json:"route_id"`
+	FailureCount   int    `json:"failure_count"`
+	LastErrorClass string `json:"last_error_class,omitempty"`
+	LastFailureAt  string `json:"last_failure_at,omitempty"`
+	ExpiresAt      string `json:"expires_at,omitempty"`
+}
+
+type RouteCooldownState struct {
+	RouteID   string `json:"route_id"`
+	Reason    string `json:"reason,omitempty"`
+	Until     string `json:"until,omitempty"`
+	ExpiresAt string `json:"expires_at,omitempty"`
+}
+
+type StickyStore interface {
+	Get(ctx context.Context, key string) (StickyBinding, bool, error)
+	Set(ctx context.Context, key string, binding StickyBinding, ttl time.Duration) error
+	Delete(ctx context.Context, key string) error
+
+	GetRouteFailure(ctx context.Context, routeID string) (RouteFailureState, bool, error)
+	SetRouteFailure(ctx context.Context, routeID string, state RouteFailureState, ttl time.Duration) error
+	ClearRouteFailure(ctx context.Context, routeID string) error
+
+	GetCooldown(ctx context.Context, routeID string) (RouteCooldownState, bool, error)
+	SetCooldown(ctx context.Context, routeID string, state RouteCooldownState, ttl time.Duration) error
+	ClearCooldown(ctx context.Context, routeID string) error
+}
+
+type StoreConfig struct {
+	Backend string
+	Redis   RedisConfig
+}
+
+func NormalizeRuntimeBackend(backend string) (string, error) {
+	switch strings.ToLower(strings.TrimSpace(backend)) {
+	case "", RuntimeBackendMemory:
+		return RuntimeBackendMemory, nil
+	case RuntimeBackendRedis:
+		return RuntimeBackendRedis, nil
+	default:
+		return "", fmt.Errorf("unsupported route runtime backend %q", backend)
+	}
+}
+
+func BuildStickyKey(scope, logicalGroupID, publicModel, subjectID string) (string, error) {
+	scope = strings.ToLower(strings.TrimSpace(scope))
+	logicalGroupID = strings.TrimSpace(logicalGroupID)
+	publicModel = strings.TrimSpace(publicModel)
+	subjectID = strings.TrimSpace(subjectID)
+
+	switch {
+	case logicalGroupID == "":
+		return "", fmt.Errorf("logical_group_id is required")
+	case publicModel == "":
+		return "", fmt.Errorf("public_model is required")
+	case subjectID == "":
+		return "", fmt.Errorf("subject_id is required")
+	}
+
+	var subjectPrefix string
+	switch scope {
+	case StickyScopeConversation:
+		subjectPrefix = "conv"
+	case StickyScopeSession:
+		subjectPrefix = "sess"
+	case StickyScopeUser:
+		subjectPrefix = "user"
+	default:
+		return "", fmt.Errorf("unsupported sticky scope %q", scope)
+	}
+
+	return fmt.Sprintf("lg:%s:m:%s:%s:%s", logicalGroupID, publicModel, subjectPrefix, subjectID), nil
+}
+
+func BuildRouteFailureKey(routeID string) (string, error) {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return "", fmt.Errorf("route_id is required")
+	}
+	return "routefail:" + routeID, nil
+}
+
+func BuildRouteCooldownKey(routeID string) (string, error) {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return "", fmt.Errorf("route_id is required")
+	}
+	return "routecool:" + routeID, nil
+}
+
+func normalizeStickyBinding(binding StickyBinding, ttl time.Duration, now time.Time) (StickyBinding, error) {
+	binding.LogicalGroupID = strings.TrimSpace(binding.LogicalGroupID)
+	binding.PublicModel = strings.TrimSpace(binding.PublicModel)
+	binding.RouteID = strings.TrimSpace(binding.RouteID)
+	binding.ShadowGroupID = strings.TrimSpace(binding.ShadowGroupID)
+
+	switch {
+	case binding.LogicalGroupID == "":
+		return StickyBinding{}, fmt.Errorf("logical_group_id is required")
+	case binding.PublicModel == "":
+		return StickyBinding{}, fmt.Errorf("public_model is required")
+	case binding.RouteID == "":
+		return StickyBinding{}, fmt.Errorf("route_id is required")
+	case binding.ShadowGroupID == "":
+		return StickyBinding{}, fmt.Errorf("shadow_group_id is required")
+	case ttl <= 0:
+		return StickyBinding{}, fmt.Errorf("ttl must be positive")
+	}
+
+	if binding.BoundAt == "" {
+		binding.BoundAt = now.UTC().Format(time.RFC3339)
+	}
+	if binding.ExpiresAt == "" {
+		binding.ExpiresAt = now.UTC().Add(ttl).Format(time.RFC3339)
+	}
+	return binding, nil
+}
+
+func normalizeRouteFailureState(routeID string, state RouteFailureState, ttl time.Duration, now time.Time) (RouteFailureState, error) {
+	routeID = strings.TrimSpace(routeID)
+	state.RouteID = strings.TrimSpace(state.RouteID)
+	state.LastErrorClass = strings.TrimSpace(state.LastErrorClass)
+
+	if state.RouteID == "" {
+		state.RouteID = routeID
+	}
+	switch {
+	case routeID == "":
+		return RouteFailureState{}, fmt.Errorf("route_id is required")
+	case state.RouteID != routeID:
+		return RouteFailureState{}, fmt.Errorf("route_id mismatch")
+	case state.FailureCount < 0:
+		return RouteFailureState{}, fmt.Errorf("failure_count must be >= 0")
+	case ttl <= 0:
+		return RouteFailureState{}, fmt.Errorf("ttl must be positive")
+	}
+
+	if state.LastFailureAt == "" {
+		state.LastFailureAt = now.UTC().Format(time.RFC3339)
+	}
+	if state.ExpiresAt == "" {
+		state.ExpiresAt = now.UTC().Add(ttl).Format(time.RFC3339)
+	}
+	return state, nil
+}
+
+func normalizeRouteCooldownState(routeID string, state RouteCooldownState, ttl time.Duration, now time.Time) (RouteCooldownState, error) {
+	routeID = strings.TrimSpace(routeID)
+	state.RouteID = strings.TrimSpace(state.RouteID)
+	state.Reason = strings.TrimSpace(state.Reason)
+
+	if state.RouteID == "" {
+		state.RouteID = routeID
+	}
+	switch {
+	case routeID == "":
+		return RouteCooldownState{}, fmt.Errorf("route_id is required")
+	case state.RouteID != routeID:
+		return RouteCooldownState{}, fmt.Errorf("route_id mismatch")
+	case ttl <= 0:
+		return RouteCooldownState{}, fmt.Errorf("ttl must be positive")
+	}
+
+	if state.Until == "" {
+		state.Until = now.UTC().Add(ttl).Format(time.RFC3339)
+	}
+	if state.ExpiresAt == "" {
+		state.ExpiresAt = now.UTC().Add(ttl).Format(time.RFC3339)
+	}
+	return state, nil
+}
--- a/internal/routing/sticky_memory.go
+++ b/internal/routing/sticky_memory.go
@@ -0,0 +1,175 @@
+package routing
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+)
+
+type memoryStickyEntry struct {
+	binding   StickyBinding
+	expiresAt time.Time
+}
+
+type memoryRouteFailureEntry struct {
+	state     RouteFailureState
+	expiresAt time.Time
+}
+
+type memoryRouteCooldownEntry struct {
+	state     RouteCooldownState
+	expiresAt time.Time
+}
+
+type InMemoryStickyStore struct {
+	now          func() time.Time
+	mu           sync.RWMutex
+	bindings     map[string]memoryStickyEntry
+	routeFailure map[string]memoryRouteFailureEntry
+	cooldowns    map[string]memoryRouteCooldownEntry
+}
+
+func NewInMemoryStickyStore() *InMemoryStickyStore {
+	return &InMemoryStickyStore{
+		now:          time.Now,
+		bindings:     make(map[string]memoryStickyEntry),
+		routeFailure: make(map[string]memoryRouteFailureEntry),
+		cooldowns:    make(map[string]memoryRouteCooldownEntry),
+	}
+}
+
+func (s *InMemoryStickyStore) Get(_ context.Context, key string) (StickyBinding, bool, error) {
+	key = strings.TrimSpace(key)
+	if key == "" {
+		return StickyBinding{}, false, nil
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	entry, ok := s.bindings[key]
+	if !ok {
+		return StickyBinding{}, false, nil
+	}
+	if s.expired(entry.expiresAt) {
+		delete(s.bindings, key)
+		return StickyBinding{}, false, nil
+	}
+	return entry.binding, true, nil
+}
+
+func (s *InMemoryStickyStore) Set(_ context.Context, key string, binding StickyBinding, ttl time.Duration) error {
+	key = strings.TrimSpace(key)
+	binding, err := normalizeStickyBinding(binding, ttl, s.now())
+	if err != nil {
+		return err
+	}
+	if key == "" {
+		return nil
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.bindings[key] = memoryStickyEntry{binding: binding, expiresAt: s.now().UTC().Add(ttl)}
+	return nil
+}
+
+func (s *InMemoryStickyStore) Delete(_ context.Context, key string) error {
+	key = strings.TrimSpace(key)
+	if key == "" {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.bindings, key)
+	return nil
+}
+
+func (s *InMemoryStickyStore) GetRouteFailure(_ context.Context, routeID string) (RouteFailureState, bool, error) {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return RouteFailureState{}, false, nil
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	entry, ok := s.routeFailure[routeID]
+	if !ok {
+		return RouteFailureState{}, false, nil
+	}
+	if s.expired(entry.expiresAt) {
+		delete(s.routeFailure, routeID)
+		return RouteFailureState{}, false, nil
+	}
+	return entry.state, true, nil
+}
+
+func (s *InMemoryStickyStore) SetRouteFailure(_ context.Context, routeID string, state RouteFailureState, ttl time.Duration) error {
+	state, err := normalizeRouteFailureState(routeID, state, ttl, s.now())
+	if err != nil {
+		return err
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.routeFailure[state.RouteID] = memoryRouteFailureEntry{state: state, expiresAt: s.now().UTC().Add(ttl)}
+	return nil
+}
+
+func (s *InMemoryStickyStore) ClearRouteFailure(_ context.Context, routeID string) error {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.routeFailure, routeID)
+	return nil
+}
+
+func (s *InMemoryStickyStore) GetCooldown(_ context.Context, routeID string) (RouteCooldownState, bool, error) {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return RouteCooldownState{}, false, nil
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	entry, ok := s.cooldowns[routeID]
+	if !ok {
+		return RouteCooldownState{}, false, nil
+	}
+	if s.expired(entry.expiresAt) {
+		delete(s.cooldowns, routeID)
+		return RouteCooldownState{}, false, nil
+	}
+	return entry.state, true, nil
+}
+
+func (s *InMemoryStickyStore) SetCooldown(_ context.Context, routeID string, state RouteCooldownState, ttl time.Duration) error {
+	state, err := normalizeRouteCooldownState(routeID, state, ttl, s.now())
+	if err != nil {
+		return err
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.cooldowns[state.RouteID] = memoryRouteCooldownEntry{state: state, expiresAt: s.now().UTC().Add(ttl)}
+	return nil
+}
+
+func (s *InMemoryStickyStore) ClearCooldown(_ context.Context, routeID string) error {
+	routeID = strings.TrimSpace(routeID)
+	if routeID == "" {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.cooldowns, routeID)
+	return nil
+}
+
+func (s *InMemoryStickyStore) expired(expiresAt time.Time) bool {
+	return !expiresAt.IsZero() && !expiresAt.After(s.now().UTC())
+}
--- a/internal/routing/sticky_redis.go
+++ b/internal/routing/sticky_redis.go
@@ -0,0 +1,352 @@
+package routing
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	defaultRedisDialTimeout = 3 * time.Second
+)
+
+type RedisConfig struct {
+	Addr        string
+	Password    string
+	DB          int
+	DialTimeout time.Duration
+}
+
+type RedisStickyStore struct {
+	cfg RedisConfig
+}
+
+func NewRedisStickyStore(ctx context.Context, cfg RedisConfig) (*RedisStickyStore, error) {
+	cfg = normalizeRedisConfig(cfg)
+	if strings.TrimSpace(cfg.Addr) == "" {
+		return nil, fmt.Errorf("redis addr is required")
+	}
+
+	store := &RedisStickyStore{cfg: cfg}
+	if err := store.ping(ctx); err != nil {
+		return nil, err
+	}
+	return store, nil
+}
+
+func (s *RedisStickyStore) Get(ctx context.Context, key string) (StickyBinding, bool, error) {
+	key = strings.TrimSpace(key)
+	if key == "" {
+		return StickyBinding{}, false, nil
+	}
+
+	payload, ok, err := s.getJSON(ctx, key)
+	if err != nil || !ok {
+		return StickyBinding{}, ok, err
+	}
+	var binding StickyBinding
+	if err := json.Unmarshal(payload, &binding); err != nil {
+		return StickyBinding{}, false, fmt.Errorf("decode sticky binding %q: %w", key, err)
+	}
+	return binding, true, nil
+}
+
+func (s *RedisStickyStore) Set(ctx context.Context, key string, binding StickyBinding, ttl time.Duration) error {
+	key = strings.TrimSpace(key)
+	if key == "" {
+		return nil
+	}
+	binding, err := normalizeStickyBinding(binding, ttl, time.Now())
+	if err != nil {
+		return err
+	}
+	return s.setJSON(ctx, key, binding, ttl)
+}
+
+func (s *RedisStickyStore) Delete(ctx context.Context, key string) error {
+	key = strings.TrimSpace(key)
+	if key == "" {
+		return nil
+	}
+	return s.delKey(ctx, key)
+}
+
+func (s *RedisStickyStore) GetRouteFailure(ctx context.Context, routeID string) (RouteFailureState, bool, error) {
+	key, err := BuildRouteFailureKey(routeID)
+	if err != nil {
+		return RouteFailureState{}, false, err
+	}
+	payload, ok, err := s.getJSON(ctx, key)
+	if err != nil || !ok {
+		return RouteFailureState{}, ok, err
+	}
+	var state RouteFailureState
+	if err := json.Unmarshal(payload, &state); err != nil {
+		return RouteFailureState{}, false, fmt.Errorf("decode route failure %q: %w", routeID, err)
+	}
+	return state, true, nil
+}
+
+func (s *RedisStickyStore) SetRouteFailure(ctx context.Context, routeID string, state RouteFailureState, ttl time.Duration) error {
+	key, err := BuildRouteFailureKey(routeID)
+	if err != nil {
+		return err
+	}
+	state, err = normalizeRouteFailureState(routeID, state, ttl, time.Now())
+	if err != nil {
+		return err
+	}
+	return s.setJSON(ctx, key, state, ttl)
+}
+
+func (s *RedisStickyStore) ClearRouteFailure(ctx context.Context, routeID string) error {
+	key, err := BuildRouteFailureKey(routeID)
+	if err != nil {
+		return err
+	}
+	return s.delKey(ctx, key)
+}
+
+func (s *RedisStickyStore) GetCooldown(ctx context.Context, routeID string) (RouteCooldownState, bool, error) {
+	key, err := BuildRouteCooldownKey(routeID)
+	if err != nil {
+		return RouteCooldownState{}, false, err
+	}
+	payload, ok, err := s.getJSON(ctx, key)
+	if err != nil || !ok {
+		return RouteCooldownState{}, ok, err
+	}
+	var state RouteCooldownState
+	if err := json.Unmarshal(payload, &state); err != nil {
+		return RouteCooldownState{}, false, fmt.Errorf("decode route cooldown %q: %w", routeID, err)
+	}
+	return state, true, nil
+}
+
+func (s *RedisStickyStore) SetCooldown(ctx context.Context, routeID string, state RouteCooldownState, ttl time.Duration) error {
+	key, err := BuildRouteCooldownKey(routeID)
+	if err != nil {
+		return err
+	}
+	state, err = normalizeRouteCooldownState(routeID, state, ttl, time.Now())
+	if err != nil {
+		return err
+	}
+	return s.setJSON(ctx, key, state, ttl)
+}
+
+func (s *RedisStickyStore) ClearCooldown(ctx context.Context, routeID string) error {
+	key, err := BuildRouteCooldownKey(routeID)
+	if err != nil {
+		return err
+	}
+	return s.delKey(ctx, key)
+}
+
+func (s *RedisStickyStore) ping(ctx context.Context) error {
+	conn, reader, err := s.open(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	if err := writeRESPArray(conn, "PING"); err != nil {
+		return fmt.Errorf("redis ping write: %w", err)
+	}
+	reply, err := readRESPValue(reader)
+	if err != nil {
+		return fmt.Errorf("redis ping read: %w", err)
+	}
+	if reply.kind != '+' || reply.stringValue != "PONG" {
+		return fmt.Errorf("redis ping unexpected response: kind=%q value=%q", reply.kind, reply.stringValue)
+	}
+	return nil
+}
+
+func (s *RedisStickyStore) getJSON(ctx context.Context, key string) ([]byte, bool, error) {
+	conn, reader, err := s.open(ctx)
+	if err != nil {
+		return nil, false, err
+	}
+	defer conn.Close()
+
+	if err := writeRESPArray(conn, "GET", key); err != nil {
+		return nil, false, fmt.Errorf("redis GET %q: write: %w", key, err)
+	}
+	reply, err := readRESPValue(reader)
+	if err != nil {
+		return nil, false, fmt.Errorf("redis GET %q: read: %w", key, err)
+	}
+	switch reply.kind {
+	case '$':
+		return []byte(reply.stringValue), true, nil
+	case '_':
+		return nil, false, nil
+	default:
+		return nil, false, fmt.Errorf("redis GET %q: unexpected response %q", key, reply.kind)
+	}
+}
+
+func (s *RedisStickyStore) setJSON(ctx context.Context, key string, payload any, ttl time.Duration) error {
+	encoded, err := json.Marshal(payload)
+	if err != nil {
+		return fmt.Errorf("marshal redis value for %q: %w", key, err)
+	}
+
+	conn, reader, err := s.open(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	seconds := int(ttl / time.Second)
+	if ttl%time.Second != 0 {
+		seconds++
+	}
+	if seconds <= 0 {
+		seconds = 1
+	}
+
+	if err := writeRESPArray(conn, "SET", key, string(encoded), "EX", strconv.Itoa(seconds)); err != nil {
+		return fmt.Errorf("redis SET %q: write: %w", key, err)
+	}
+	reply, err := readRESPValue(reader)
+	if err != nil {
+		return fmt.Errorf("redis SET %q: read: %w", key, err)
+	}
+	if reply.kind != '+' || reply.stringValue != "OK" {
+		return fmt.Errorf("redis SET %q: unexpected response kind=%q value=%q", key, reply.kind, reply.stringValue)
+	}
+	return nil
+}
+
+func (s *RedisStickyStore) delKey(ctx context.Context, key string) error {
+	conn, reader, err := s.open(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	if err := writeRESPArray(conn, "DEL", key); err != nil {
+		return fmt.Errorf("redis DEL %q: write: %w", key, err)
+	}
+	reply, err := readRESPValue(reader)
+	if err != nil {
+		return fmt.Errorf("redis DEL %q: read: %w", key, err)
+	}
+	if reply.kind != ':' {
+		return fmt.Errorf("redis DEL %q: unexpected response kind=%q value=%q", key, reply.kind, reply.stringValue)
+	}
+	return nil
+}
+
+func (s *RedisStickyStore) open(ctx context.Context) (net.Conn, *bufio.Reader, error) {
+	cfg := normalizeRedisConfig(s.cfg)
+	dialer := &net.Dialer{Timeout: cfg.DialTimeout}
+	conn, err := dialer.DialContext(ctx, "tcp", cfg.Addr)
+	if err != nil {
+		return nil, nil, fmt.Errorf("dial redis %q: %w", cfg.Addr, err)
+	}
+	reader := bufio.NewReader(conn)
+
+	if strings.TrimSpace(cfg.Password) != "" {
+		if err := writeRESPArray(conn, "AUTH", cfg.Password); err != nil {
+			conn.Close()
+			return nil, nil, fmt.Errorf("redis AUTH write: %w", err)
+		}
+		reply, err := readRESPValue(reader)
+		if err != nil {
+			conn.Close()
+			return nil, nil, fmt.Errorf("redis AUTH read: %w", err)
+		}
+		if reply.kind != '+' || reply.stringValue != "OK" {
+			conn.Close()
+			return nil, nil, fmt.Errorf("redis AUTH unexpected response kind=%q value=%q", reply.kind, reply.stringValue)
+		}
+	}
+
+	if cfg.DB > 0 {
+		if err := writeRESPArray(conn, "SELECT", strconv.Itoa(cfg.DB)); err != nil {
+			conn.Close()
+			return nil, nil, fmt.Errorf("redis SELECT write: %w", err)
+		}
+		reply, err := readRESPValue(reader)
+		if err != nil {
+			conn.Close()
+			return nil, nil, fmt.Errorf("redis SELECT read: %w", err)
+		}
+		if reply.kind != '+' || reply.stringValue != "OK" {
+			conn.Close()
+			return nil, nil, fmt.Errorf("redis SELECT unexpected response kind=%q value=%q", reply.kind, reply.stringValue)
+		}
+	}
+
+	return conn, reader, nil
+}
+
+func normalizeRedisConfig(cfg RedisConfig) RedisConfig {
+	cfg.Addr = strings.TrimSpace(cfg.Addr)
+	cfg.Password = strings.TrimSpace(cfg.Password)
+	if cfg.DialTimeout <= 0 {
+		cfg.DialTimeout = defaultRedisDialTimeout
+	}
+	return cfg
+}
+
+type respValue struct {
+	kind        byte
+	stringValue string
+}
+
+func writeRESPArray(w io.Writer, parts ...string) error {
+	if _, err := io.WriteString(w, fmt.Sprintf("*%d\r\n", len(parts))); err != nil {
+		return err
+	}
+	for _, part := range parts {
+		if _, err := io.WriteString(w, fmt.Sprintf("$%d\r\n%s\r\n", len(part), part)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func readRESPValue(r *bufio.Reader) (respValue, error) {
+	prefix, err := r.ReadByte()
+	if err != nil {
+		return respValue{}, err
+	}
+
+	line, err := r.ReadString('\n')
+	if err != nil {
+		return respValue{}, err
+	}
+	line = strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r")
+
+	switch prefix {
+	case '+', '-', ':':
+		if prefix == '-' {
+			return respValue{}, fmt.Errorf("redis error: %s", line)
+		}
+		return respValue{kind: prefix, stringValue: line}, nil
+	case '$':
+		size, err := strconv.Atoi(line)
+		if err != nil {
+			return respValue{}, fmt.Errorf("parse bulk length: %w", err)
+		}
+		if size < 0 {
+			return respValue{kind: '_'}, nil
+		}
+		payload := make([]byte, size+2)
+		if _, err := io.ReadFull(r, payload); err != nil {
+			return respValue{}, err
+		}
+		return respValue{kind: '$', stringValue: string(payload[:size])}, nil
+	default:
+		return respValue{}, fmt.Errorf("unsupported redis response prefix %q", prefix)
+	}
+}
--- a/internal/routing/sticky_test.go
+++ b/internal/routing/sticky_test.go
@@ -0,0 +1,460 @@
+package routing
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestBuildStickyKey(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		scope   string
+		want    string
+		wantErr bool
+	}{
+		{name: "conversation", scope: StickyScopeConversation, want: "lg:gpt-shared:m:gpt-5.4:conv:conversation-1"},
+		{name: "session", scope: StickyScopeSession, want: "lg:gpt-shared:m:gpt-5.4:sess:session-1"},
+		{name: "user", scope: StickyScopeUser, want: "lg:gpt-shared:m:gpt-5.4:user:user-1"},
+		{name: "invalid", scope: "bad", wantErr: true},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			got, err := BuildStickyKey(tt.scope, "gpt-shared", "gpt-5.4", tt.name+"-1")
+			if tt.wantErr {
+				if err == nil {
+					t.Fatal("BuildStickyKey() error = nil, want error")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("BuildStickyKey() error = %v", err)
+			}
+			if got != tt.want {
+				t.Fatalf("BuildStickyKey() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestInMemoryStickyStoreBindingFailureAndCooldown(t *testing.T) {
+	t.Parallel()
+
+	store := NewInMemoryStickyStore()
+	ctx := context.Background()
+	key, err := BuildStickyKey(StickyScopeConversation, "gpt-shared", "gpt-5.4", "conv-1")
+	if err != nil {
+		t.Fatalf("BuildStickyKey() error = %v", err)
+	}
+
+	if err := store.Set(ctx, key, StickyBinding{
+		LogicalGroupID: "gpt-shared",
+		PublicModel:    "gpt-5.4",
+		RouteID:        "asxs",
+		ShadowGroupID:  "gpt-shared__asxs",
+	}, 2*time.Second); err != nil {
+		t.Fatalf("Set() error = %v", err)
+	}
+	binding, ok, err := store.Get(ctx, key)
+	if err != nil || !ok {
+		t.Fatalf("Get() = (%+v, %v, %v), want binding", binding, ok, err)
+	}
+	if binding.RouteID != "asxs" {
+		t.Fatalf("binding.RouteID = %q, want asxs", binding.RouteID)
+	}
+	if err := store.Delete(ctx, key); err != nil {
+		t.Fatalf("Delete() error = %v", err)
+	}
+	if _, ok, err := store.Get(ctx, key); err != nil || ok {
+		t.Fatalf("Get() after delete = (ok=%v, err=%v), want false nil", ok, err)
+	}
+
+	if err := store.SetRouteFailure(ctx, "asxs", RouteFailureState{
+		FailureCount:   2,
+		LastErrorClass: "timeout",
+	}, time.Second); err != nil {
+		t.Fatalf("SetRouteFailure() error = %v", err)
+	}
+	failure, ok, err := store.GetRouteFailure(ctx, "asxs")
+	if err != nil || !ok || failure.FailureCount != 2 {
+		t.Fatalf("GetRouteFailure() = (%+v, %v, %v), want count 2", failure, ok, err)
+	}
+	if err := store.ClearRouteFailure(ctx, "asxs"); err != nil {
+		t.Fatalf("ClearRouteFailure() error = %v", err)
+	}
+
+	if err := store.SetCooldown(ctx, "asxs", RouteCooldownState{
+		Reason: "cooldown",
+	}, time.Second); err != nil {
+		t.Fatalf("SetCooldown() error = %v", err)
+	}
+	cooldown, ok, err := store.GetCooldown(ctx, "asxs")
+	if err != nil || !ok || cooldown.RouteID != "asxs" {
+		t.Fatalf("GetCooldown() = (%+v, %v, %v), want route asxs", cooldown, ok, err)
+	}
+	if err := store.ClearCooldown(ctx, "asxs"); err != nil {
+		t.Fatalf("ClearCooldown() error = %v", err)
+	}
+}
+
+func TestInMemoryStickyStoreTTlExpiry(t *testing.T) {
+	t.Parallel()
+
+	store := NewInMemoryStickyStore()
+	ctx := context.Background()
+	key, err := BuildStickyKey(StickyScopeUser, "gpt-shared", "gpt-5.4", "user-1")
+	if err != nil {
+		t.Fatalf("BuildStickyKey() error = %v", err)
+	}
+	if err := store.Set(ctx, key, StickyBinding{
+		LogicalGroupID: "gpt-shared",
+		PublicModel:    "gpt-5.4",
+		RouteID:        "asxs",
+		ShadowGroupID:  "gpt-shared__asxs",
+	}, 40*time.Millisecond); err != nil {
+		t.Fatalf("Set() error = %v", err)
+	}
+	time.Sleep(60 * time.Millisecond)
+	if _, ok, err := store.Get(ctx, key); err != nil || ok {
+		t.Fatalf("Get() after ttl = (ok=%v, err=%v), want false nil", ok, err)
+	}
+}
+
+func TestRedisStickyStoreRoundTripWithFakeServer(t *testing.T) {
+	t.Parallel()
+
+	server := newFakeRedisServer(t)
+	defer server.Close()
+
+	store, err := NewRedisStickyStore(context.Background(), RedisConfig{
+		Addr:     server.Addr(),
+		Password: "secret",
+		DB:       2,
+	})
+	if err != nil {
+		t.Fatalf("NewRedisStickyStore() error = %v", err)
+	}
+
+	ctx := context.Background()
+	key, err := BuildStickyKey(StickyScopeSession, "gpt-shared", "gpt-5.4", "sess-1")
+	if err != nil {
+		t.Fatalf("BuildStickyKey() error = %v", err)
+	}
+	if err := store.Set(ctx, key, StickyBinding{
+		LogicalGroupID: "gpt-shared",
+		PublicModel:    "gpt-5.4",
+		RouteID:        "asxs",
+		ShadowGroupID:  "gpt-shared__asxs",
+	}, time.Minute); err != nil {
+		t.Fatalf("Set() error = %v", err)
+	}
+	if binding, ok, err := store.Get(ctx, key); err != nil || !ok || binding.RouteID != "asxs" {
+		t.Fatalf("Get() = (%+v, %v, %v), want route asxs", binding, ok, err)
+	}
+	if err := store.SetRouteFailure(ctx, "asxs", RouteFailureState{
+		FailureCount:   3,
+		LastErrorClass: "timeout",
+	}, time.Minute); err != nil {
+		t.Fatalf("SetRouteFailure() error = %v", err)
+	}
+	if state, ok, err := store.GetRouteFailure(ctx, "asxs"); err != nil || !ok || state.FailureCount != 3 {
+		t.Fatalf("GetRouteFailure() = (%+v, %v, %v), want count 3", state, ok, err)
+	}
+	if err := store.SetCooldown(ctx, "asxs", RouteCooldownState{
+		Reason: "degraded",
+	}, time.Minute); err != nil {
+		t.Fatalf("SetCooldown() error = %v", err)
+	}
+	if state, ok, err := store.GetCooldown(ctx, "asxs"); err != nil || !ok || state.Reason != "degraded" {
+		t.Fatalf("GetCooldown() = (%+v, %v, %v), want reason degraded", state, ok, err)
+	}
+	if err := store.Delete(ctx, key); err != nil {
+		t.Fatalf("Delete() error = %v", err)
+	}
+	if _, ok, err := store.Get(ctx, key); err != nil || ok {
+		t.Fatalf("Get() after delete = (ok=%v, err=%v), want false nil", ok, err)
+	}
+}
+
+type fakeRedisServer struct {
+	t        *testing.T
+	listener net.Listener
+	password string
+	mu       sync.Mutex
+	values   map[int]map[string]fakeRedisValue
+}
+
+type fakeRedisValue struct {
+	value     string
+	expiresAt time.Time
+}
+
+func newFakeRedisServer(t *testing.T) *fakeRedisServer {
+	t.Helper()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("net.Listen() error = %v", err)
+	}
+	server := &fakeRedisServer{
+		t:        t,
+		listener: ln,
+		password: "secret",
+		values:   make(map[int]map[string]fakeRedisValue),
+	}
+	go server.serve()
+	return server
+}
+
+func (s *fakeRedisServer) Addr() string {
+	return s.listener.Addr().String()
+}
+
+func (s *fakeRedisServer) Close() {
+	_ = s.listener.Close()
+}
+
+func (s *fakeRedisServer) serve() {
+	for {
+		conn, err := s.listener.Accept()
+		if err != nil {
+			return
+		}
+		go s.handleConn(conn)
+	}
+}
+
+func (s *fakeRedisServer) handleConn(conn net.Conn) {
+	defer conn.Close()
+
+	reader := bufio.NewReader(conn)
+	currentDB := 0
+	authed := false
+	for {
+		command, err := readRESPArray(reader)
+		if err != nil {
+			if err == io.EOF {
+				return
+			}
+			s.writeError(conn, err.Error())
+			return
+		}
+		if len(command) == 0 {
+			s.writeError(conn, "empty command")
+			continue
+		}
+
+		switch strings.ToUpper(command[0]) {
+		case "PING":
+			s.writeSimpleString(conn, "PONG")
+		case "AUTH":
+			if len(command) != 2 || command[1] != s.password {
+				s.writeError(conn, "ERR invalid password")
+				continue
+			}
+			authed = true
+			s.writeSimpleString(conn, "OK")
+		case "SELECT":
+			if len(command) != 2 {
+				s.writeError(conn, "ERR bad select")
+				continue
+			}
+			db, err := strconv.Atoi(command[1])
+			if err != nil {
+				s.writeError(conn, "ERR bad db")
+				continue
+			}
+			currentDB = db
+			s.writeSimpleString(conn, "OK")
+		case "SET":
+			if !authed {
+				s.writeError(conn, "NOAUTH")
+				continue
+			}
+			if len(command) != 5 || strings.ToUpper(command[3]) != "EX" {
+				s.writeError(conn, "ERR bad set")
+				continue
+			}
+			ttl, err := strconv.Atoi(command[4])
+			if err != nil {
+				s.writeError(conn, "ERR bad ttl")
+				continue
+			}
+			s.setValue(currentDB, command[1], command[2], time.Duration(ttl)*time.Second)
+			s.writeSimpleString(conn, "OK")
+		case "GET":
+			if !authed {
+				s.writeError(conn, "NOAUTH")
+				continue
+			}
+			if len(command) != 2 {
+				s.writeError(conn, "ERR bad get")
+				continue
+			}
+			value, ok := s.getValue(currentDB, command[1])
+			if !ok {
+				s.writeNullBulk(conn)
+				continue
+			}
+			s.writeBulk(conn, value)
+		case "DEL":
+			if !authed {
+				s.writeError(conn, "NOAUTH")
+				continue
+			}
+			if len(command) != 2 {
+				s.writeError(conn, "ERR bad del")
+				continue
+			}
+			s.deleteValue(currentDB, command[1])
+			s.writeInteger(conn, 1)
+		default:
+			s.writeError(conn, "ERR unknown command")
+		}
+	}
+}
+
+func readRESPArray(reader *bufio.Reader) ([]string, error) {
+	line, err := reader.ReadString('\n')
+	if err != nil {
+		return nil, err
+	}
+	line = strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r")
+	if !strings.HasPrefix(line, "*") {
+		return nil, fmt.Errorf("expected array, got %q", line)
+	}
+	count, err := strconv.Atoi(strings.TrimPrefix(line, "*"))
+	if err != nil {
+		return nil, err
+	}
+
+	parts := make([]string, 0, count)
+	for i := 0; i < count; i++ {
+		header, err := reader.ReadString('\n')
+		if err != nil {
+			return nil, err
+		}
+		header = strings.TrimSuffix(strings.TrimSuffix(header, "\n"), "\r")
+		if !strings.HasPrefix(header, "$") {
+			return nil, fmt.Errorf("expected bulk header, got %q", header)
+		}
+		size, err := strconv.Atoi(strings.TrimPrefix(header, "$"))
+		if err != nil {
+			return nil, err
+		}
+		payload := make([]byte, size+2)
+		if _, err := io.ReadFull(reader, payload); err != nil {
+			return nil, err
+		}
+		parts = append(parts, string(payload[:size]))
+	}
+	return parts, nil
+}
+
+func (s *fakeRedisServer) writeSimpleString(w io.Writer, value string) {
+	_, _ = io.WriteString(w, "+"+value+"\r\n")
+}
+
+func (s *fakeRedisServer) writeBulk(w io.Writer, value string) {
+	_, _ = io.WriteString(w, fmt.Sprintf("$%d\r\n%s\r\n", len(value), value))
+}
+
+func (s *fakeRedisServer) writeNullBulk(w io.Writer) {
+	_, _ = io.WriteString(w, "$-1\r\n")
+}
+
+func (s *fakeRedisServer) writeInteger(w io.Writer, value int) {
+	_, _ = io.WriteString(w, fmt.Sprintf(":%d\r\n", value))
+}
+
+func (s *fakeRedisServer) writeError(w io.Writer, message string) {
+	_, _ = io.WriteString(w, "-"+message+"\r\n")
+}
+
+func (s *fakeRedisServer) setValue(db int, key, value string, ttl time.Duration) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.values[db] == nil {
+		s.values[db] = make(map[string]fakeRedisValue)
+	}
+	s.values[db][key] = fakeRedisValue{value: value, expiresAt: time.Now().Add(ttl)}
+}
+
+func (s *fakeRedisServer) getValue(db int, key string) (string, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	value, ok := s.values[db][key]
+	if !ok {
+		return "", false
+	}
+	if !value.expiresAt.IsZero() && !value.expiresAt.After(time.Now()) {
+		delete(s.values[db], key)
+		return "", false
+	}
+	return value.value, true
+}
+
+func (s *fakeRedisServer) deleteValue(db int, key string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.values[db] == nil {
+		return
+	}
+	delete(s.values[db], key)
+}
+
+func TestRedisStickyStoreRequiresAddr(t *testing.T) {
+	t.Parallel()
+
+	if _, err := NewRedisStickyStore(context.Background(), RedisConfig{}); err == nil {
+		t.Fatal("NewRedisStickyStore() error = nil, want missing addr")
+	}
+}
+
+func TestNormalizeRuntimeBackend(t *testing.T) {
+	t.Parallel()
+
+	if got, err := NormalizeRuntimeBackend(""); err != nil || got != RuntimeBackendMemory {
+		t.Fatalf("NormalizeRuntimeBackend(\"\") = (%q, %v), want memory nil", got, err)
+	}
+	if got, err := NormalizeRuntimeBackend("redis"); err != nil || got != RuntimeBackendRedis {
+		t.Fatalf("NormalizeRuntimeBackend(redis) = (%q, %v), want redis nil", got, err)
+	}
+	if _, err := NormalizeRuntimeBackend("bad"); err == nil {
+		t.Fatal("NormalizeRuntimeBackend(bad) error = nil, want error")
+	}
+}
+
+func TestRouteFailureAndCooldownKeyBuilders(t *testing.T) {
+	t.Parallel()
+
+	failureKey, err := BuildRouteFailureKey("asxs")
+	if err != nil || failureKey != "routefail:asxs" {
+		t.Fatalf("BuildRouteFailureKey() = (%q, %v), want routefail:asxs nil", failureKey, err)
+	}
+	cooldownKey, err := BuildRouteCooldownKey("asxs")
+	if err != nil || cooldownKey != "routecool:asxs" {
+		t.Fatalf("BuildRouteCooldownKey() = (%q, %v), want routecool:asxs nil", cooldownKey, err)
+	}
+}
+
+func TestRedisStickyStoreFixturePathExists(t *testing.T) {
+	t.Parallel()
+
+	if filepath.Base(t.TempDir()) == "" {
+		t.Fatal("temp dir base should not be empty")
+	}
+}