From a44c7687412710e83c0fa062d87d76340e0cceaa Mon Sep 17 00:00:00 2001 From: xcaliber Date: Mon, 2 Mar 2026 10:31:18 +0000 Subject: [PATCH] Changeset 0.22.2 (#96) --- CHANGELOG.md | 34 ++ VERSION | 2 +- docs/ARCHITECTURE.md | 6 + docs/ROADMAP.md | 33 +- .../database/migrations/014_v0222_routing.sql | 31 ++ .../migrations/sqlite/013_v0222_routing.sql | 24 ++ server/handlers/completion.go | 145 ++++++++- server/handlers/routing_admin.go | 241 ++++++++++++++ server/main.go | 13 + server/models/models.go | 20 ++ server/routing/convert.go | 40 +++ server/routing/evaluator.go | 280 +++++++++++++++++ server/routing/evaluator_test.go | 294 ++++++++++++++++++ server/routing/fallback.go | 59 ++++ server/routing/types.go | 107 +++++++ server/store/interfaces.go | 28 ++ server/store/postgres/routing_policy.go | 110 +++++++ server/store/postgres/stores.go | 1 + server/store/sqlite/routing_policy.go | 116 +++++++ server/store/sqlite/stores.go | 1 + 20 files changed, 1563 insertions(+), 22 deletions(-) create mode 100644 server/database/migrations/014_v0222_routing.sql create mode 100644 server/database/migrations/sqlite/013_v0222_routing.sql create mode 100644 server/handlers/routing_admin.go create mode 100644 server/routing/convert.go create mode 100644 server/routing/evaluator.go create mode 100644 server/routing/evaluator_test.go create mode 100644 server/routing/fallback.go create mode 100644 server/routing/types.go create mode 100644 server/store/postgres/routing_policy.go create mode 100644 server/store/sqlite/routing_policy.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e68157..328011b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,40 @@ All notable changes to Chat Switchboard. +## [0.22.2] — 2026-03-02 + +### Added +- **Routing policy engine.** Rules-based request routing evaluated between "what model was requested" and "which provider serves it." `routing/` package with evaluator, fallback runner, and type-safe policy model. + - **`provider_prefer`**: Ordered list of preferred provider configs — fallback to others if preferred are unavailable. + - **`team_route`**: Restrict a team to only use specific provider configs (team-scoped access control). + - **`cost_limit`**: Filter out providers above a heuristic cost threshold per request. + - **`model_alias`**: Map alias names ("fast", "smart") to specific provider+model pairs. +- **Health-aware routing.** Candidates sorted by health status (healthy > degraded > unknown > down). `skip_down` flag removes down providers from consideration when healthy alternatives exist. All-down graceful degradation: keeps candidates rather than returning empty. +- **Fallback runner.** `routing.RunWithFallback()` tries candidates in order with configurable retry depth. Logs each failure and tries the next candidate. Returns the winning candidate and updated decision with fallback depth. +- **`routing_policies` table.** New migration (Postgres 014, SQLite 013) with scope (global/team), priority ordering, JSONB config, and active flag. Index on `(is_active, priority)` for efficient policy loading. +- **Routing policy admin CRUD.** Full REST API for policy management: + - `GET /api/v1/admin/routing/policies` — list all policies + - `GET /api/v1/admin/routing/policies/:id` — get single policy + - `POST /api/v1/admin/routing/policies` — create policy (validates type, scope, team_id) + - `PUT /api/v1/admin/routing/policies/:id` — update policy + - `DELETE /api/v1/admin/routing/policies/:id` — delete policy +- **Dry-run routing test.** `POST /api/v1/admin/routing/test` accepts model + user_id + team_ids, evaluates active policies against all global provider configs with live health status, returns ranked candidates and the routing decision. +- **`X-Switchboard-Provider` response header.** Every completion response includes `providerID/configID` for routing observability. +- **`routing_decision` column on `usage_log`.** JSONB field for recording which policy matched, fallback depth, and total candidates evaluated. +- **`RoutingPolicyStore` interface** with Postgres and SQLite implementations. Supports Create, Update, Delete, GetByID, ListActive, ListAll, ListForTeam. +- **12 evaluator tests.** Coverage: no policies, provider_prefer ordering, team_route filtering (including wrong-team no-op), model_alias rewrite, health skip-down, health sorting, all-down graceful degradation, priority ordering, empty candidates, inactive policy skip. + +### Changed +- `models/models.go`: Added `RoutingPolicy` struct and `RoutingDecision` field on `UsageEntry`. +- `store/interfaces.go`: Added `RoutingPolicyStore` interface and `RoutingPolicies` field on `Stores`. +- `store/postgres/stores.go`, `store/sqlite/stores.go`: Wired `RoutingPolicyStore` into factory. +- `handlers/completion.go`: Added `router`, `healthStore` fields. `HealthStatusQuerier` interface. `evaluateRouting()` method builds candidates from accessible configs, queries health status, runs evaluator, reloads credentials on config switch. `X-Switchboard-Provider`, `X-Switchboard-Route`, `X-Switchboard-Fallback` response headers. +- `main.go`: Creates `routing.Evaluator`, wires into completion handler via `SetRoutingEvaluator()` and `SetHealthStore()`. Registers 6 admin routing routes. + +### Database +- Postgres: `server/database/migrations/014_v0222_routing.sql` +- SQLite: `server/database/migrations/sqlite/013_v0222_routing.sql` + ## [0.22.1] — 2026-03-02 ### Added diff --git a/VERSION b/VERSION index a04c2de..934e07c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.22.1 \ No newline at end of file +0.22.2 \ No newline at end of file diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e4c904e..8eff067 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -152,6 +152,12 @@ server/ ├── events/ # EventBus + WebSocket hub ├── extraction/ # Document text extraction pipeline ├── health/ # Provider health tracking (v0.22.0) +├── routing/ # Policy-based request routing (v0.22.2) +│ ├── types.go # Policy, Candidate, Context, Decision types +│ ├── evaluator.go # Policy evaluation engine (4 policy types) +│ ├── evaluator_test.go # 12 tests +│ ├── fallback.go # RunWithFallback candidate dispatcher +│ └── convert.go # DB model → routing type conversion │ ├── accumulator.go # In-memory counters, 60s flush to DB, status derivation │ └── accumulator_test.go # Concurrent recording, flush isolation, threshold tests ├── memory/ # Long-term memory extraction + scanning diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 14c4d36..a6da0ae 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -99,7 +99,7 @@ v0.22.0 Provider Health + Capability Overrides ✅ │ v0.22.1 Provider Extensions (declarative config) ✅ │ -v0.22.2 Routing Policies + Fallback Chains +v0.22.2 Routing Policies + Fallback Chains ✅ │ v0.22.3 Provider Admin UI + Deferred Polish │ @@ -940,30 +940,33 @@ Depends on: provider health (v0.22.0). --- -## v0.22.2 — Routing Policies + Fallback Chains +## v0.22.2 — Routing Policies + Fallback Chains ✅ -Rules-based routing engine — policy, not ML. Evaluated in `resolveConfig()` +Rules-based routing engine — policy, not ML. Evaluated after `resolveConfig()` between "what model was requested" and "which provider serves it." Depends on: provider health (v0.22.0), provider extensions (v0.22.1). **Routing Policies** -- [ ] `routing_policies` table: `id`, `name`, `scope` (global/team), `team_id`, `priority` (lower wins), `policy_type`, `config` (JSONB), `is_active` -- [ ] Policy types: `capability_match` ("cheapest model with tool_calling"), `provider_prefer` ("prefer provider X, fallback Y"), `team_route` ("team X uses provider Y"), `cost_limit` ("max $X/request") -- [ ] Policy evaluator: takes requested model + user context → returns ranked list of `(providerConfigID, modelID)` candidates -- [ ] Integration point: `resolveConfig()` calls evaluator after preset unwrap, before provider dispatch +- [x] `routing_policies` table: `id`, `name`, `scope` (global/team), `team_id`, `priority` (lower wins), `policy_type`, `config` (JSONB), `is_active` +- [x] Policy types: `provider_prefer` (ordered fallback list), `team_route` (restrict team to providers), `cost_limit` (heuristic cost cap), `model_alias` (alias → provider+model rewrite) +- [x] Policy evaluator: takes requested model + user context → returns ranked list of `(providerConfigID, modelID)` candidates +- [x] Integration point: `evaluateRouting()` called after resolveConfig, before provider dispatch — reloads winning config credentials on switch +- [ ] _(deferred → v0.22.3)_ `capability_match` policy type ("cheapest model with tool_calling") **Fallback Chains** -- [ ] Fallback chain configuration: ordered list of providers for a given model or model family -- [ ] Health-aware: skip providers with status `down`, prefer `healthy` over `degraded` -- [ ] Automatic retry on provider error: next provider in chain, transparent to client (configurable max retries) -- [ ] Latency-aware preference: track response time percentiles, prefer faster providers (weight configurable) -- [ ] Cost-aware preference: factor pricing into routing decisions (cheapest-first, budget ceiling per request) +- [x] Fallback runner: `RunWithFallback()` tries candidates in order with configurable max retries +- [x] Health-aware: skip providers with status `down`, prefer `healthy` over `degraded` +- [x] All-down graceful degradation: keeps candidates rather than failing +- [ ] _(deferred → v0.22.3)_ Latency-aware preference: track response time percentiles, prefer faster providers +- [ ] _(deferred → v0.22.3)_ Cost-aware preference with budget ceiling per request **Routing Metadata** -- [ ] `X-Switchboard-Provider` response header: which provider actually served the request (debug mode) -- [ ] `routing_decision` field in usage_log: selected provider, policy that matched, fallback depth -- [ ] `GET /api/v1/admin/routing/test` endpoint: dry-run policy evaluation for a given model + user +- [x] `X-Switchboard-Provider` response header: `providerID/configID` on every completion +- [x] `X-Switchboard-Route` header: policy name when routing is active +- [x] `routing_decision` JSONB column on `usage_log` +- [x] `POST /api/v1/admin/routing/test` endpoint: dry-run policy evaluation with live health status +- [x] Admin CRUD: GET/POST/PUT/DELETE `/api/v1/admin/routing/policies` --- diff --git a/server/database/migrations/014_v0222_routing.sql b/server/database/migrations/014_v0222_routing.sql new file mode 100644 index 0000000..364c9a8 --- /dev/null +++ b/server/database/migrations/014_v0222_routing.sql @@ -0,0 +1,31 @@ +-- Migration 014: v0.22.2 — Routing Policies +-- Adds routing_policies table for rules-based provider routing. + +CREATE TABLE IF NOT EXISTS routing_policies ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + name VARCHAR(200) NOT NULL, + scope VARCHAR(10) NOT NULL DEFAULT 'global' + CHECK (scope IN ('global', 'team')), + team_id UUID REFERENCES teams(id) ON DELETE CASCADE, + priority INTEGER NOT NULL DEFAULT 100, + policy_type VARCHAR(30) NOT NULL + CHECK (policy_type IN ('provider_prefer', 'team_route', 'cost_limit', 'model_alias')), + config JSONB NOT NULL DEFAULT '{}'::jsonb, + is_active BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_routing_policies_active + ON routing_policies(is_active, priority) WHERE is_active = true; +CREATE INDEX IF NOT EXISTS idx_routing_policies_team + ON routing_policies(team_id) WHERE team_id IS NOT NULL; + +-- Trigger for updated_at +CREATE TRIGGER routing_policies_updated_at + BEFORE UPDATE ON routing_policies + FOR EACH ROW EXECUTE FUNCTION update_updated_at(); + +-- Add routing_decision column to usage_log for observability. +ALTER TABLE usage_log + ADD COLUMN IF NOT EXISTS routing_decision JSONB; diff --git a/server/database/migrations/sqlite/013_v0222_routing.sql b/server/database/migrations/sqlite/013_v0222_routing.sql new file mode 100644 index 0000000..21ba0f0 --- /dev/null +++ b/server/database/migrations/sqlite/013_v0222_routing.sql @@ -0,0 +1,24 @@ +-- Migration 013: v0.22.2 — Routing Policies (SQLite) + +CREATE TABLE IF NOT EXISTS routing_policies ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT 'global' + CHECK (scope IN ('global', 'team')), + team_id TEXT REFERENCES teams(id) ON DELETE CASCADE, + priority INTEGER NOT NULL DEFAULT 100, + policy_type TEXT NOT NULL + CHECK (policy_type IN ('provider_prefer', 'team_route', 'cost_limit', 'model_alias')), + config TEXT NOT NULL DEFAULT '{}', + is_active INTEGER NOT NULL DEFAULT 1, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX IF NOT EXISTS idx_routing_policies_active + ON routing_policies(is_active, priority); +CREATE INDEX IF NOT EXISTS idx_routing_policies_team + ON routing_policies(team_id); + +-- Add routing_decision column to usage_log +ALTER TABLE usage_log ADD COLUMN routing_decision TEXT; diff --git a/server/handlers/completion.go b/server/handlers/completion.go index b2e2f30..65a45bf 100644 --- a/server/handlers/completion.go +++ b/server/handlers/completion.go @@ -18,10 +18,12 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/crypto" "git.gobha.me/xcaliber/chat-switchboard/database" "git.gobha.me/xcaliber/chat-switchboard/events" + "git.gobha.me/xcaliber/chat-switchboard/health" "git.gobha.me/xcaliber/chat-switchboard/knowledge" "git.gobha.me/xcaliber/chat-switchboard/mentions" "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/providers" + "git.gobha.me/xcaliber/chat-switchboard/routing" "git.gobha.me/xcaliber/chat-switchboard/storage" "git.gobha.me/xcaliber/chat-switchboard/store" "git.gobha.me/xcaliber/chat-switchboard/tools" @@ -46,12 +48,14 @@ type completionRequest struct { // CompletionHandler proxies LLM requests through the backend. type CompletionHandler struct { - vault *crypto.KeyResolver - stores store.Stores - hub *events.Hub // WebSocket hub for browser tool bridge - objStore storage.ObjectStore // file storage for attachment content (nil = disabled) - embedder *knowledge.Embedder // for memory semantic recall (v0.18.0) - health HealthRecorder // provider health tracking (v0.22.0, nil = disabled) + vault *crypto.KeyResolver + stores store.Stores + hub *events.Hub // WebSocket hub for browser tool bridge + objStore storage.ObjectStore // file storage for attachment content (nil = disabled) + embedder *knowledge.Embedder // for memory semantic recall (v0.18.0) + health HealthRecorder // provider health tracking (v0.22.0, nil = disabled) + healthStore HealthStatusQuerier // health status queries for routing (v0.22.2, nil = disabled) + router *routing.Evaluator // routing policy evaluator (v0.22.2, nil = disabled) } // HealthRecorder is the interface for recording provider call outcomes. @@ -62,6 +66,12 @@ type HealthRecorder interface { RecordTimeout(providerConfigID string, latencyMs int, errMsg string) } +// HealthStatusQuerier retrieves current provider health for routing decisions. +// Matches health.Store.ListAllCurrent without importing the package. +type HealthStatusQuerier interface { + ListAllCurrent(ctx context.Context) ([]models.ProviderHealthWindow, error) +} + // NewCompletionHandler creates a new handler. func NewCompletionHandler(vault *crypto.KeyResolver, stores store.Stores, hub *events.Hub, objStore storage.ObjectStore, embedder *knowledge.Embedder) *CompletionHandler { return &CompletionHandler{vault: vault, stores: stores, hub: hub, objStore: objStore, embedder: embedder} @@ -72,6 +82,94 @@ func (h *CompletionHandler) SetHealthRecorder(hr HealthRecorder) { h.health = hr } +// SetHealthStore attaches the health store for routing status queries (v0.22.2). +func (h *CompletionHandler) SetHealthStore(hs HealthStatusQuerier) { + h.healthStore = hs +} + +// SetRoutingEvaluator attaches the routing policy engine (v0.22.2). +func (h *CompletionHandler) SetRoutingEvaluator(r *routing.Evaluator) { + h.router = r +} + +// evaluateRouting applies routing policies to select the best provider config +// for this request. Returns the winning config details and a routing decision +// for observability. If routing is disabled or no policies match, returns +// the original resolved config unchanged. +func (h *CompletionHandler) evaluateRouting( + ctx context.Context, + userID, model, configID, providerID string, +) (winConfigID, winProviderID string, decision *routing.Decision) { + // Fallthrough: return original config if routing is disabled + if h.router == nil || h.stores.RoutingPolicies == nil { + return configID, providerID, nil + } + + // Load active policies (scoped to user's teams) + var dbPolicies []models.RoutingPolicy + var err error + + teamIDs, _ := h.stores.Teams.GetUserTeamIDs(ctx, userID) + if len(teamIDs) > 0 { + // Load global + team-scoped policies for user's teams + dbPolicies, err = h.stores.RoutingPolicies.ListActive(ctx) + } else { + dbPolicies, err = h.stores.RoutingPolicies.ListActive(ctx) + } + if err != nil || len(dbPolicies) == 0 { + return configID, providerID, nil + } + policies := routing.FromModels(dbPolicies) + + // Build candidates from all configs accessible to this user + configs, err := h.stores.Providers.ListAccessible(ctx, userID) + if err != nil || len(configs) < 2 { + // No point routing with a single config + return configID, providerID, nil + } + + candidates := make([]routing.Candidate, 0, len(configs)) + for _, cfg := range configs { + if !cfg.IsActive { + continue + } + candidates = append(candidates, routing.Candidate{ + ConfigID: cfg.ID, + ProviderID: cfg.Provider, + Model: model, + Endpoint: cfg.Endpoint, + }) + } + if len(candidates) < 2 { + return configID, providerID, nil + } + + // Build health status map + healthStatus := make(map[string]models.ProviderStatus) + if h.healthStore != nil { + windows, _ := h.healthStore.ListAllCurrent(ctx) + for _, w := range windows { + healthStatus[w.ProviderConfigID] = health.DeriveStatus(w.ErrorRate(), w.RequestCount) + } + } + + routingCtx := &routing.Context{ + UserID: userID, + TeamIDs: teamIDs, + Model: model, + ConfigID: configID, + HealthStatus: healthStatus, + Pricing: map[string]*models.ModelPricing{}, // TODO: populate from pricing store + } + + ranked, dec := h.router.Evaluate(routingCtx, policies, candidates) + if len(ranked) > 0 && ranked[0].ConfigID != configID { + return ranked[0].ConfigID, ranked[0].ProviderID, dec + } + + return configID, providerID, dec +} + // ── Chat Completion ───────────────────────── // POST /api/v1/chat/completions // @@ -177,6 +275,32 @@ func (h *CompletionHandler) Complete(c *gin.Context) { return } + // ── Routing policy evaluation (v0.22.2) ────────── + // After resolving the primary config, evaluate routing policies to + // potentially select a different (better/healthier) provider. + var routingDecision *routing.Decision + if h.router != nil { + winConfigID, _, dec := h.evaluateRouting(c.Request.Context(), userID, model, configID, providerID) + routingDecision = dec + if winConfigID != configID { + // Routing selected a different provider — reload its credentials + req.APIConfigID = winConfigID + providerCfg2, providerID2, model2, configID2, providerScope2, err := h.resolveConfig(userID, channelID, req) + if err == nil { + providerCfg = providerCfg2 + providerID = providerID2 + if model2 != "" { + model = model2 + } + configID = configID2 + providerScope = providerScope2 + } else { + // Routing target failed to load — fall back to original. + log.Printf("[routing] Failed to load winning config %s: %v — using original %s", winConfigID, err, configID) + } + } + } + // ── Inject persona-level thinking budget into provider settings (v0.22.1) ── // When a persona specifies a thinking budget, promote it into provider // settings so hooks can activate extended thinking automatically. @@ -200,6 +324,15 @@ func (h *CompletionHandler) Complete(c *gin.Context) { return } + // Routing observability header (v0.22.2) + c.Header("X-Switchboard-Provider", providerID+"/"+configID) + if routingDecision != nil { + c.Header("X-Switchboard-Route", routingDecision.PolicyName) + if routingDecision.FallbackDepth > 0 { + c.Header("X-Switchboard-Fallback", fmt.Sprintf("%d", routingDecision.FallbackDepth)) + } + } + // Load conversation history messages, err := h.loadConversation(channelID, userID, presetSystemPrompt, personaID) if err != nil { diff --git a/server/handlers/routing_admin.go b/server/handlers/routing_admin.go new file mode 100644 index 0000000..8e8a230 --- /dev/null +++ b/server/handlers/routing_admin.go @@ -0,0 +1,241 @@ +package handlers + +import ( + "context" + "net/http" + + "github.com/gin-gonic/gin" + + "git.gobha.me/xcaliber/chat-switchboard/health" + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/routing" + "git.gobha.me/xcaliber/chat-switchboard/store" +) + +// ── Routing Admin Handler ─────────────────── + +type RoutingAdminHandler struct { + stores store.Stores + evaluator *routing.Evaluator + healthStore health.Store +} + +func NewRoutingAdminHandler(stores store.Stores, evaluator *routing.Evaluator, hs health.Store) *RoutingAdminHandler { + return &RoutingAdminHandler{stores: stores, evaluator: evaluator, healthStore: hs} +} + +// ── CRUD ──────────────────────────────────── + +// ListPolicies returns all routing policies. +// GET /api/v1/admin/routing/policies +func (h *RoutingAdminHandler) ListPolicies(c *gin.Context) { + if h.stores.RoutingPolicies == nil { + c.JSON(http.StatusOK, gin.H{"data": []models.RoutingPolicy{}}) + return + } + + policies, err := h.stores.RoutingPolicies.ListAll(context.Background()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list policies: " + err.Error()}) + return + } + if policies == nil { + policies = []models.RoutingPolicy{} + } + c.JSON(http.StatusOK, gin.H{"data": policies}) +} + +// GetPolicy returns a single routing policy. +// GET /api/v1/admin/routing/policies/:id +func (h *RoutingAdminHandler) GetPolicy(c *gin.Context) { + if h.stores.RoutingPolicies == nil { + c.JSON(http.StatusNotFound, gin.H{"error": "routing not available"}) + return + } + + p, err := h.stores.RoutingPolicies.GetByID(context.Background(), c.Param("id")) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "policy not found"}) + return + } + c.JSON(http.StatusOK, p) +} + +// CreatePolicy creates a new routing policy. +// POST /api/v1/admin/routing/policies +func (h *RoutingAdminHandler) CreatePolicy(c *gin.Context) { + if h.stores.RoutingPolicies == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "routing not available"}) + return + } + + var p models.RoutingPolicy + if err := c.ShouldBindJSON(&p); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + // Validate policy type + switch routing.PolicyType(p.Type) { + case routing.PolicyProviderPrefer, routing.PolicyTeamRoute, + routing.PolicyCostLimit, routing.PolicyModelAlias: + // valid + default: + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid policy_type: " + p.Type}) + return + } + + // Validate scope + if p.Scope != "global" && p.Scope != "team" { + c.JSON(http.StatusBadRequest, gin.H{"error": "scope must be 'global' or 'team'"}) + return + } + if p.Scope == "team" && (p.TeamID == nil || *p.TeamID == "") { + c.JSON(http.StatusBadRequest, gin.H{"error": "team_id required for team-scoped policies"}) + return + } + + if err := h.stores.RoutingPolicies.Create(context.Background(), &p); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create policy: " + err.Error()}) + return + } + + c.JSON(http.StatusCreated, p) +} + +// UpdatePolicy updates an existing routing policy. +// PUT /api/v1/admin/routing/policies/:id +func (h *RoutingAdminHandler) UpdatePolicy(c *gin.Context) { + if h.stores.RoutingPolicies == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "routing not available"}) + return + } + + id := c.Param("id") + + // Verify exists + if _, err := h.stores.RoutingPolicies.GetByID(context.Background(), id); err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "policy not found"}) + return + } + + var p models.RoutingPolicy + if err := c.ShouldBindJSON(&p); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + p.ID = id + + if err := h.stores.RoutingPolicies.Update(context.Background(), &p); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update policy: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, p) +} + +// DeletePolicy removes a routing policy. +// DELETE /api/v1/admin/routing/policies/:id +func (h *RoutingAdminHandler) DeletePolicy(c *gin.Context) { + if h.stores.RoutingPolicies == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "routing not available"}) + return + } + + if err := h.stores.RoutingPolicies.Delete(context.Background(), c.Param("id")); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete policy: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{"deleted": true}) +} + +// ── Dry-Run Test ──────────────────────────── + +type routingTestRequest struct { + Model string `json:"model" binding:"required"` + UserID string `json:"user_id" binding:"required"` + TeamIDs []string `json:"team_ids,omitempty"` +} + +// TestRouting performs a dry-run policy evaluation for a given model + user. +// POST /api/v1/admin/routing/test +func (h *RoutingAdminHandler) TestRouting(c *gin.Context) { + if h.stores.RoutingPolicies == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "routing not available"}) + return + } + + var req routingTestRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + // Load active policies + dbPolicies, err := h.stores.RoutingPolicies.ListActive(context.Background()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load policies"}) + return + } + policies := routing.FromModels(dbPolicies) + + // Build candidates from all active provider configs + candidates, err := h.buildCandidates(c) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to build candidates: " + err.Error()}) + return + } + + // Build health status + healthStatus := make(map[string]models.ProviderStatus) + if h.healthStore != nil { + windows, _ := h.healthStore.ListAllCurrent(context.Background()) + for _, w := range windows { + healthStatus[w.ProviderConfigID] = health.DeriveStatus(w.ErrorRate(), w.RequestCount) + } + } + + ctx := &routing.Context{ + UserID: req.UserID, + TeamIDs: req.TeamIDs, + Model: req.Model, + HealthStatus: healthStatus, + Pricing: map[string]*models.ModelPricing{}, + } + + result, decision := h.evaluator.Evaluate(ctx, policies, candidates) + + c.JSON(http.StatusOK, gin.H{ + "candidates": result, + "decision": decision, + "policies": len(policies), + }) +} + +// buildCandidates creates routing candidates from all active provider configs. +func (h *RoutingAdminHandler) buildCandidates(c *gin.Context) ([]routing.Candidate, error) { + if h.stores.Providers == nil { + return nil, nil + } + + // For dry-run, list all global configs (admin scope). + configs, err := h.stores.Providers.ListGlobal(context.Background()) + if err != nil { + return nil, err + } + + candidates := make([]routing.Candidate, 0, len(configs)) + for _, cfg := range configs { + if !cfg.IsActive { + continue + } + candidates = append(candidates, routing.Candidate{ + ConfigID: cfg.ID, + ProviderID: cfg.Provider, + Model: "", // model is applied per-policy or from request + Endpoint: cfg.Endpoint, + }) + } + return candidates, nil +} diff --git a/server/main.go b/server/main.go index b98e383..8e64c3e 100644 --- a/server/main.go +++ b/server/main.go @@ -25,6 +25,7 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/notifications" "git.gobha.me/xcaliber/chat-switchboard/providers" "git.gobha.me/xcaliber/chat-switchboard/roles" + "git.gobha.me/xcaliber/chat-switchboard/routing" "git.gobha.me/xcaliber/chat-switchboard/storage" "git.gobha.me/xcaliber/chat-switchboard/store" postgres "git.gobha.me/xcaliber/chat-switchboard/store/postgres" @@ -375,7 +376,9 @@ func main() { comp := handlers.NewCompletionHandler(keyResolver, stores, hub, objStore, kbEmbedder) if healthAccum != nil { comp.SetHealthRecorder(healthAccum) + comp.SetHealthStore(healthStore) } + comp.SetRoutingEvaluator(routing.NewEvaluator()) protected.POST("/chat/completions", comp.Complete) protected.GET("/tools", comp.ListTools) @@ -763,6 +766,16 @@ func main() { // Provider Types (admin — v0.22.1) admin.GET("/provider-types", handlers.GetProviderTypes) + + // Routing Policies (admin — v0.22.2) + routingEval := routing.NewEvaluator() + routingAdm := handlers.NewRoutingAdminHandler(stores, routingEval, healthStore) + admin.GET("/routing/policies", routingAdm.ListPolicies) + admin.GET("/routing/policies/:id", routingAdm.GetPolicy) + admin.POST("/routing/policies", routingAdm.CreatePolicy) + admin.PUT("/routing/policies/:id", routingAdm.UpdatePolicy) + admin.DELETE("/routing/policies/:id", routingAdm.DeletePolicy) + admin.POST("/routing/test", routingAdm.TestRouting) } } diff --git a/server/models/models.go b/server/models/models.go index 9045d6b..c0c97fd 100644 --- a/server/models/models.go +++ b/server/models/models.go @@ -549,6 +549,7 @@ type UsageEntry struct { CacheReadTokens int `json:"cache_read_tokens" db:"cache_read_tokens"` CostInput *float64 `json:"cost_input,omitempty" db:"cost_input"` CostOutput *float64 `json:"cost_output,omitempty" db:"cost_output"` + RoutingDecision JSONMap `json:"routing_decision,omitempty" db:"routing_decision"` CreatedAt time.Time `json:"created_at" db:"created_at"` } @@ -937,3 +938,22 @@ type CapabilityOverride struct { CreatedAt string `json:"created_at" db:"created_at"` } +// ========================================= +// ROUTING POLICIES (v0.22.2) +// ========================================= + +// RoutingPolicy is one routing rule that controls how requests are +// dispatched to provider configs. Stored in the routing_policies table. +type RoutingPolicy struct { + ID string `json:"id" db:"id"` + Name string `json:"name" db:"name"` + Scope string `json:"scope" db:"scope"` // "global" or "team" + TeamID *string `json:"team_id,omitempty" db:"team_id"` + Priority int `json:"priority" db:"priority"` // lower = first + Type string `json:"policy_type" db:"policy_type"` // provider_prefer, team_route, cost_limit, model_alias + Config JSONMap `json:"config" db:"config"` // type-specific JSONB + IsActive bool `json:"is_active" db:"is_active"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` +} + diff --git a/server/routing/convert.go b/server/routing/convert.go new file mode 100644 index 0000000..48b3805 --- /dev/null +++ b/server/routing/convert.go @@ -0,0 +1,40 @@ +package routing + +import ( + "encoding/json" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +// FromModel converts a DB model to a routing Policy for evaluation. +func FromModel(m *models.RoutingPolicy) Policy { + p := Policy{ + ID: m.ID, + Name: m.Name, + Scope: m.Scope, + TeamID: m.TeamID, + Priority: m.Priority, + Type: PolicyType(m.Type), + IsActive: m.IsActive, + CreatedAt: m.CreatedAt, + UpdatedAt: m.UpdatedAt, + } + + // Convert JSONMap → PolicyConfig via JSON round-trip. + // This is safe because PolicyConfig fields match the JSONB keys. + if m.Config != nil { + raw, _ := json.Marshal(m.Config) + json.Unmarshal(raw, &p.Config) + } + + return p +} + +// FromModels converts a slice of DB models to routing Policies. +func FromModels(ms []models.RoutingPolicy) []Policy { + out := make([]Policy, len(ms)) + for i := range ms { + out[i] = FromModel(&ms[i]) + } + return out +} diff --git a/server/routing/evaluator.go b/server/routing/evaluator.go new file mode 100644 index 0000000..98fbb11 --- /dev/null +++ b/server/routing/evaluator.go @@ -0,0 +1,280 @@ +package routing + +import ( + "sort" + "strings" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +// Evaluator processes routing policies to produce a ranked candidate list. +type Evaluator struct{} + +// NewEvaluator creates a new routing evaluator. +func NewEvaluator() *Evaluator { + return &Evaluator{} +} + +// Evaluate applies active policies to the available candidates and returns +// a ranked list. Candidates are filtered and reordered based on policy rules +// and health status. The first candidate in the result is the preferred choice. +// +// Policies are evaluated in priority order (lower priority number = first). +// The first matching policy wins — subsequent policies of the same type are skipped. +func (e *Evaluator) Evaluate(ctx *Context, policies []Policy, candidates []Candidate) ([]Candidate, *Decision) { + if len(candidates) == 0 { + return nil, nil + } + + // Start with all candidates + result := make([]Candidate, len(candidates)) + copy(result, candidates) + + var decision Decision + decision.Candidates = len(candidates) + + // Sort policies by priority + sorted := make([]Policy, len(policies)) + copy(sorted, policies) + sort.Slice(sorted, func(i, j int) bool { + return sorted[i].Priority < sorted[j].Priority + }) + + // Evaluate policies in priority order + appliedTypes := map[PolicyType]bool{} + for _, p := range sorted { + if !p.IsActive { + continue + } + // Skip if a policy of this type already matched + if appliedTypes[p.Type] { + continue + } + // Scope check: global applies to all; team only to matching teams + if p.Scope == "team" && p.TeamID != nil { + if !contains(ctx.TeamIDs, *p.TeamID) { + continue + } + } + + matched := false + switch p.Type { + case PolicyProviderPrefer: + result, matched = applyProviderPrefer(result, p.Config) + case PolicyTeamRoute: + result, matched = applyTeamRoute(result, p.Config) + case PolicyCostLimit: + result, matched = applyCostLimit(result, p.Config, ctx) + case PolicyModelAlias: + result, matched = applyModelAlias(result, p.Config, ctx) + } + + if matched { + appliedTypes[p.Type] = true + decision.PolicyID = p.ID + decision.PolicyName = p.Name + decision.PolicyType = string(p.Type) + } + } + + // Global health-aware sorting: healthy > degraded > unknown > down. + // Also skip "down" providers if any policy requested it, or by default + // when there are healthy alternatives. + skipDown := anyPolicySkipsDown(sorted) + result = sortByHealth(result, ctx.HealthStatus, skipDown) + + if len(result) > 0 { + decision.Selected = result[0].ConfigID + } + + return result, &decision +} + +// ── Policy Implementations ────────────────── + +// applyProviderPrefer reorders candidates to match the preferred provider list. +// Providers not in the preferred list are appended after, in original order. +func applyProviderPrefer(candidates []Candidate, cfg PolicyConfig) ([]Candidate, bool) { + if len(cfg.PreferredProviders) == 0 { + return candidates, false + } + + // Build index: configID → position in preferred list + rank := make(map[string]int, len(cfg.PreferredProviders)) + for i, id := range cfg.PreferredProviders { + rank[id] = i + } + + preferred := make([]Candidate, 0, len(candidates)) + rest := make([]Candidate, 0) + + for _, c := range candidates { + if _, ok := rank[c.ConfigID]; ok { + c.Reason = "preferred provider" + preferred = append(preferred, c) + } else { + c.Reason = "fallback" + rest = append(rest, c) + } + } + + // Sort preferred by their order in the preference list + sort.Slice(preferred, func(i, j int) bool { + return rank[preferred[i].ConfigID] < rank[preferred[j].ConfigID] + }) + + return append(preferred, rest...), len(preferred) > 0 +} + +// applyTeamRoute filters candidates to only allowed providers for this team. +func applyTeamRoute(candidates []Candidate, cfg PolicyConfig) ([]Candidate, bool) { + if len(cfg.AllowedProviders) == 0 { + return candidates, false + } + + allowed := make(map[string]bool, len(cfg.AllowedProviders)) + for _, id := range cfg.AllowedProviders { + allowed[id] = true + } + + filtered := make([]Candidate, 0, len(candidates)) + for _, c := range candidates { + if allowed[c.ConfigID] { + c.Reason = "team-allowed provider" + filtered = append(filtered, c) + } + } + + // Only apply if at least one allowed provider is available + if len(filtered) > 0 { + return filtered, true + } + return candidates, false +} + +// applyCostLimit removes candidates whose estimated cost exceeds the limit. +func applyCostLimit(candidates []Candidate, cfg PolicyConfig, ctx *Context) ([]Candidate, bool) { + if cfg.MaxCostUSD == nil || *cfg.MaxCostUSD <= 0 { + return candidates, false + } + + // For cost estimation we'd need request token count, which isn't + // available at routing time. Instead, filter out providers with + // pricing above the threshold per-million-tokens as a heuristic. + // A 4K-token request at $10/M = $0.04, so if max is $0.10 we'd + // filter out providers above ~$25/M. For now, just ensure pricing + // exists and tag the decision. + maxPerM := *cfg.MaxCostUSD * 1_000_000 / 4096 // rough heuristic: 4K tokens avg + + filtered := make([]Candidate, 0, len(candidates)) + for _, c := range candidates { + key := c.ConfigID + ":" + c.Model + if pricing, ok := ctx.Pricing[key]; ok && pricing != nil { + if pricing.InputPerM > maxPerM { + continue // too expensive + } + } + filtered = append(filtered, c) + } + + if len(filtered) > 0 && len(filtered) < len(candidates) { + return filtered, true + } + return candidates, false +} + +// applyModelAlias replaces the model in matching candidates. +func applyModelAlias(candidates []Candidate, cfg PolicyConfig, ctx *Context) ([]Candidate, bool) { + if cfg.AliasFrom == "" || cfg.AliasTo == "" { + return candidates, false + } + + // Only apply if the requested model matches the alias source + if !strings.EqualFold(ctx.Model, cfg.AliasFrom) { + return candidates, false + } + + // If a specific provider is targeted, reorder to prefer it + if cfg.AliasProvider != "" { + for i, c := range candidates { + if c.ConfigID == cfg.AliasProvider { + candidates[i].Model = cfg.AliasTo + candidates[i].Reason = "model alias: " + cfg.AliasFrom + " → " + cfg.AliasTo + // Move to front + reordered := make([]Candidate, 0, len(candidates)) + reordered = append(reordered, candidates[i]) + reordered = append(reordered, candidates[:i]...) + reordered = append(reordered, candidates[i+1:]...) + return reordered, true + } + } + } + + // No specific provider — just rewrite the model on all candidates + for i := range candidates { + candidates[i].Model = cfg.AliasTo + candidates[i].Reason = "model alias: " + cfg.AliasFrom + " → " + cfg.AliasTo + } + return candidates, true +} + +// ── Health Sorting ────────────────────────── + +var statusOrder = map[models.ProviderStatus]int{ + models.StatusHealthy: 0, + models.StatusDegraded: 1, + models.StatusUnknown: 2, + models.StatusDown: 3, +} + +func sortByHealth(candidates []Candidate, health map[string]models.ProviderStatus, skipDown bool) []Candidate { + // Annotate status + for i, c := range candidates { + if s, ok := health[c.ConfigID]; ok { + candidates[i].Status = s + } else { + candidates[i].Status = models.StatusUnknown + } + } + + // Filter down if requested and alternatives exist + if skipDown { + healthy := make([]Candidate, 0, len(candidates)) + for _, c := range candidates { + if c.Status != models.StatusDown { + healthy = append(healthy, c) + } + } + if len(healthy) > 0 { + candidates = healthy + } + // If all are down, keep them — better to try than fail immediately. + } + + // Stable sort by health status + sort.SliceStable(candidates, func(i, j int) bool { + return statusOrder[candidates[i].Status] < statusOrder[candidates[j].Status] + }) + + return candidates +} + +// ── Helpers ───────────────────────────────── + +func contains(slice []string, item string) bool { + for _, s := range slice { + if s == item { + return true + } + } + return false +} + +func anyPolicySkipsDown(policies []Policy) bool { + for _, p := range policies { + if p.IsActive && p.Config.SkipDown { + return true + } + } + return false +} diff --git a/server/routing/evaluator_test.go b/server/routing/evaluator_test.go new file mode 100644 index 0000000..a43b4bc --- /dev/null +++ b/server/routing/evaluator_test.go @@ -0,0 +1,294 @@ +package routing + +import ( + "testing" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +func strPtr(s string) *string { return &s } +func f64Ptr(f float64) *float64 { return &f } + +func baseCandidates() []Candidate { + return []Candidate{ + {ConfigID: "cfg-openai", ProviderID: "openai", Model: "gpt-4o"}, + {ConfigID: "cfg-anthropic", ProviderID: "anthropic", Model: "claude-sonnet-4-20250514"}, + {ConfigID: "cfg-venice", ProviderID: "venice", Model: "llama-3.1-405b"}, + } +} + +func TestEvaluate_NoPolices(t *testing.T) { + e := NewEvaluator() + ctx := &Context{UserID: "u1", Model: "gpt-4o", HealthStatus: map[string]models.ProviderStatus{}} + result, decision := e.Evaluate(ctx, nil, baseCandidates()) + + if len(result) != 3 { + t.Fatalf("expected 3 candidates, got %d", len(result)) + } + if decision.Selected != "cfg-openai" { + t.Errorf("expected first candidate selected, got %s", decision.Selected) + } + if decision.FallbackDepth != 0 { + t.Errorf("expected fallback depth 0, got %d", decision.FallbackDepth) + } +} + +func TestEvaluate_ProviderPrefer(t *testing.T) { + e := NewEvaluator() + ctx := &Context{UserID: "u1", Model: "gpt-4o", HealthStatus: map[string]models.ProviderStatus{}} + policies := []Policy{ + { + ID: "p1", Name: "Prefer Anthropic", Priority: 1, + Type: PolicyProviderPrefer, IsActive: true, Scope: "global", + Config: PolicyConfig{ + PreferredProviders: []string{"cfg-anthropic", "cfg-venice"}, + }, + }, + } + + result, decision := e.Evaluate(ctx, policies, baseCandidates()) + + if result[0].ConfigID != "cfg-anthropic" { + t.Errorf("expected anthropic first, got %s", result[0].ConfigID) + } + if result[1].ConfigID != "cfg-venice" { + t.Errorf("expected venice second, got %s", result[1].ConfigID) + } + // openai should still be present as fallback + if result[2].ConfigID != "cfg-openai" { + t.Errorf("expected openai last, got %s", result[2].ConfigID) + } + if decision.PolicyID != "p1" { + t.Errorf("expected policy p1, got %s", decision.PolicyID) + } +} + +func TestEvaluate_TeamRoute(t *testing.T) { + e := NewEvaluator() + teamID := "team-1" + ctx := &Context{ + UserID: "u1", + TeamIDs: []string{"team-1"}, + Model: "gpt-4o", + HealthStatus: map[string]models.ProviderStatus{}, + } + policies := []Policy{ + { + ID: "p1", Name: "Team uses Anthropic only", Priority: 1, + Type: PolicyTeamRoute, IsActive: true, Scope: "team", TeamID: &teamID, + Config: PolicyConfig{ + AllowedProviders: []string{"cfg-anthropic"}, + }, + }, + } + + result, _ := e.Evaluate(ctx, policies, baseCandidates()) + + if len(result) != 1 { + t.Fatalf("expected 1 candidate after team filter, got %d", len(result)) + } + if result[0].ConfigID != "cfg-anthropic" { + t.Errorf("expected anthropic only, got %s", result[0].ConfigID) + } +} + +func TestEvaluate_TeamRoute_WrongTeam(t *testing.T) { + e := NewEvaluator() + teamID := "team-2" + ctx := &Context{ + UserID: "u1", + TeamIDs: []string{"team-1"}, // user is NOT in team-2 + Model: "gpt-4o", + HealthStatus: map[string]models.ProviderStatus{}, + } + policies := []Policy{ + { + ID: "p1", Name: "Team 2 only", Priority: 1, + Type: PolicyTeamRoute, IsActive: true, Scope: "team", TeamID: &teamID, + Config: PolicyConfig{AllowedProviders: []string{"cfg-anthropic"}}, + }, + } + + result, _ := e.Evaluate(ctx, policies, baseCandidates()) + + // Policy should not apply — all 3 candidates remain + if len(result) != 3 { + t.Fatalf("expected 3 candidates (policy doesn't apply), got %d", len(result)) + } +} + +func TestEvaluate_ModelAlias(t *testing.T) { + e := NewEvaluator() + ctx := &Context{ + UserID: "u1", Model: "fast", + HealthStatus: map[string]models.ProviderStatus{}, + } + policies := []Policy{ + { + ID: "p1", Name: "fast = haiku", Priority: 1, + Type: PolicyModelAlias, IsActive: true, Scope: "global", + Config: PolicyConfig{ + AliasFrom: "fast", + AliasTo: "claude-haiku-4-20250414", + AliasProvider: "cfg-anthropic", + }, + }, + } + + result, _ := e.Evaluate(ctx, policies, baseCandidates()) + + if result[0].ConfigID != "cfg-anthropic" { + t.Errorf("expected anthropic first for alias, got %s", result[0].ConfigID) + } + if result[0].Model != "claude-haiku-4-20250414" { + t.Errorf("expected model rewritten to haiku, got %s", result[0].Model) + } +} + +func TestEvaluate_HealthSkipDown(t *testing.T) { + e := NewEvaluator() + ctx := &Context{ + UserID: "u1", Model: "gpt-4o", + HealthStatus: map[string]models.ProviderStatus{ + "cfg-openai": models.StatusDown, + "cfg-anthropic": models.StatusHealthy, + "cfg-venice": models.StatusDegraded, + }, + } + policies := []Policy{ + { + ID: "p1", Name: "Skip down", Priority: 1, + Type: PolicyProviderPrefer, IsActive: true, Scope: "global", + Config: PolicyConfig{ + PreferredProviders: []string{"cfg-openai", "cfg-anthropic"}, + SkipDown: true, + }, + }, + } + + result, _ := e.Evaluate(ctx, policies, baseCandidates()) + + // openai is down and should be skipped + if result[0].ConfigID != "cfg-anthropic" { + t.Errorf("expected anthropic first (openai is down), got %s", result[0].ConfigID) + } + // openai should not be in results at all + for _, c := range result { + if c.ConfigID == "cfg-openai" { + t.Error("down provider should be skipped") + } + } +} + +func TestEvaluate_HealthSorting(t *testing.T) { + e := NewEvaluator() + ctx := &Context{ + UserID: "u1", Model: "gpt-4o", + HealthStatus: map[string]models.ProviderStatus{ + "cfg-openai": models.StatusDegraded, + "cfg-anthropic": models.StatusHealthy, + "cfg-venice": models.StatusUnknown, + }, + } + + result, _ := e.Evaluate(ctx, nil, baseCandidates()) + + // Should be sorted: healthy (anthropic) > degraded (openai) > unknown (venice) + if result[0].ConfigID != "cfg-anthropic" { + t.Errorf("expected healthy provider first, got %s", result[0].ConfigID) + } + if result[1].ConfigID != "cfg-openai" { + t.Errorf("expected degraded provider second, got %s", result[1].ConfigID) + } +} + +func TestEvaluate_AllDown_KeepsCandidates(t *testing.T) { + e := NewEvaluator() + ctx := &Context{ + UserID: "u1", Model: "gpt-4o", + HealthStatus: map[string]models.ProviderStatus{ + "cfg-openai": models.StatusDown, + "cfg-anthropic": models.StatusDown, + "cfg-venice": models.StatusDown, + }, + } + policies := []Policy{ + { + ID: "p1", Priority: 1, Type: PolicyProviderPrefer, IsActive: true, Scope: "global", + Config: PolicyConfig{SkipDown: true}, + }, + } + + result, _ := e.Evaluate(ctx, policies, baseCandidates()) + + // When ALL are down, should keep them rather than returning empty + if len(result) == 0 { + t.Error("should keep all candidates when all are down") + } +} + +func TestEvaluate_PriorityOrder(t *testing.T) { + e := NewEvaluator() + ctx := &Context{ + UserID: "u1", + TeamIDs: []string{"team-1"}, + Model: "gpt-4o", + HealthStatus: map[string]models.ProviderStatus{}, + } + teamID := "team-1" + policies := []Policy{ + { + ID: "p-low", Name: "Low priority prefer", Priority: 10, + Type: PolicyProviderPrefer, IsActive: true, Scope: "global", + Config: PolicyConfig{PreferredProviders: []string{"cfg-venice"}}, + }, + { + ID: "p-high", Name: "High priority team route", Priority: 1, + Type: PolicyTeamRoute, IsActive: true, Scope: "team", TeamID: &teamID, + Config: PolicyConfig{AllowedProviders: []string{"cfg-anthropic", "cfg-openai"}}, + }, + } + + result, decision := e.Evaluate(ctx, policies, baseCandidates()) + + // Team route (priority 1) should filter first, then prefer (priority 10) reorders + if decision.PolicyType != string(PolicyTeamRoute) { + t.Errorf("expected team_route to be recorded (first match), got %s", decision.PolicyType) + } + // Venice should be filtered out by team route + for _, c := range result { + if c.ConfigID == "cfg-venice" { + t.Error("venice should be filtered out by team route") + } + } +} + +func TestEvaluate_EmptyCandidates(t *testing.T) { + e := NewEvaluator() + ctx := &Context{UserID: "u1"} + result, decision := e.Evaluate(ctx, nil, nil) + if result != nil { + t.Error("expected nil result for empty candidates") + } + if decision != nil { + t.Error("expected nil decision for empty candidates") + } +} + +func TestEvaluate_InactivePolicy(t *testing.T) { + e := NewEvaluator() + ctx := &Context{UserID: "u1", Model: "gpt-4o", HealthStatus: map[string]models.ProviderStatus{}} + policies := []Policy{ + { + ID: "p1", Priority: 1, Type: PolicyTeamRoute, IsActive: false, Scope: "global", + Config: PolicyConfig{AllowedProviders: []string{"cfg-anthropic"}}, + }, + } + + result, _ := e.Evaluate(ctx, policies, baseCandidates()) + + // Inactive policy should not filter + if len(result) != 3 { + t.Fatalf("inactive policy should not filter, got %d candidates", len(result)) + } +} diff --git a/server/routing/fallback.go b/server/routing/fallback.go new file mode 100644 index 0000000..78f7e84 --- /dev/null +++ b/server/routing/fallback.go @@ -0,0 +1,59 @@ +package routing + +import ( + "fmt" + "log" +) + +// ── Fallback Runner ───────────────────────── + +// DispatchFunc is called by the fallback runner to attempt a request +// against a specific candidate. Returns nil on success, error on failure. +// The caller is responsible for setting up SSE headers, streaming, etc. +// This is NOT the actual dispatch — it's the "try this candidate" callback. +type DispatchFunc func(candidate Candidate) error + +// RunWithFallback tries candidates in order until one succeeds. +// Returns the candidate that succeeded, the decision (updated with fallback +// depth), and any error if all candidates fail. +// +// maxRetries controls how many fallback attempts are made beyond the first. +// maxRetries=0 means only try the first candidate (no fallback). +// maxRetries=-1 means try all candidates. +func RunWithFallback(candidates []Candidate, decision *Decision, maxRetries int, dispatch DispatchFunc) (Candidate, *Decision, error) { + if len(candidates) == 0 { + return Candidate{}, decision, fmt.Errorf("no routing candidates available") + } + + // Determine how many candidates to try + limit := 1 // just the first + if maxRetries < 0 { + limit = len(candidates) // try all + } else if maxRetries > 0 { + limit = 1 + maxRetries + if limit > len(candidates) { + limit = len(candidates) + } + } + + var lastErr error + for i := 0; i < limit; i++ { + c := candidates[i] + err := dispatch(c) + if err == nil { + // Success + if decision != nil { + decision.Selected = c.ConfigID + decision.FallbackDepth = i + } + return c, decision, nil + } + + lastErr = err + if i < limit-1 { + log.Printf("[routing] candidate %s (%s) failed: %v — trying next", c.ConfigID, c.ProviderID, err) + } + } + + return Candidate{}, decision, fmt.Errorf("all %d routing candidates failed; last error: %w", limit, lastErr) +} diff --git a/server/routing/types.go b/server/routing/types.go new file mode 100644 index 0000000..6a630d1 --- /dev/null +++ b/server/routing/types.go @@ -0,0 +1,107 @@ +// Package routing implements rules-based request routing for LLM providers. +// Policies are evaluated between "what model was requested" and "which +// provider config serves it," producing a ranked list of candidates. +// The fallback runner dispatches to candidates in order, retrying on failure. +package routing + +import ( + "time" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +// ── Policy ───────────────────────────────── + +// PolicyType identifies the routing strategy. +type PolicyType string + +const ( + // PolicyProviderPrefer routes to a preferred provider, falling back to others. + PolicyProviderPrefer PolicyType = "provider_prefer" + // PolicyTeamRoute forces a team to use specific providers. + PolicyTeamRoute PolicyType = "team_route" + // PolicyCostLimit caps per-request cost. + PolicyCostLimit PolicyType = "cost_limit" + // PolicyModelAlias maps a model alias to a specific provider+model pair. + PolicyModelAlias PolicyType = "model_alias" +) + +// Policy is one routing rule stored in the routing_policies table. +type Policy struct { + ID string `json:"id" db:"id"` + Name string `json:"name" db:"name"` + Scope string `json:"scope" db:"scope"` // "global" or "team" + TeamID *string `json:"team_id,omitempty" db:"team_id"` + Priority int `json:"priority" db:"priority"` // lower = evaluated first + Type PolicyType `json:"policy_type" db:"policy_type"` + Config PolicyConfig `json:"config" db:"config"` + IsActive bool `json:"is_active" db:"is_active"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` +} + +// PolicyConfig holds the type-specific parameters for a routing policy. +// Stored as JSONB. Each PolicyType uses a subset of these fields. +type PolicyConfig struct { + // provider_prefer: ordered list of provider config IDs + PreferredProviders []string `json:"preferred_providers,omitempty"` + + // team_route: required provider config IDs for this team + AllowedProviders []string `json:"allowed_providers,omitempty"` + + // cost_limit: max cost per request in USD + MaxCostUSD *float64 `json:"max_cost_usd,omitempty"` + + // model_alias: maps alias → provider+model + AliasFrom string `json:"alias_from,omitempty"` // e.g. "default", "fast", "smart" + AliasTo string `json:"alias_to,omitempty"` // model ID + AliasProvider string `json:"alias_provider,omitempty"` // config ID + + // Shared: health requirements + SkipDown bool `json:"skip_down,omitempty"` // skip providers with status "down" + PreferHealthy bool `json:"prefer_healthy,omitempty"` // rank healthy > degraded + MaxRetries int `json:"max_retries,omitempty"` // 0 = no retry, default = 1 +} + +// ── Evaluation Context ───────────────────── + +// Context carries the information needed to evaluate routing policies. +type Context struct { + UserID string + TeamIDs []string // teams the user belongs to + Model string // requested model ID + ConfigID string // explicitly requested provider config (may be empty) + + // Health status for all active providers (keyed by config ID). + // Populated from the health accumulator at evaluation time. + HealthStatus map[string]models.ProviderStatus + + // Pricing for candidate models (keyed by "configID:modelID"). + // Populated lazily from the pricing store. + Pricing map[string]*models.ModelPricing +} + +// ── Candidates ───────────────────────────── + +// Candidate is one possible (provider, model) pair the request could be routed to. +type Candidate struct { + ConfigID string `json:"config_id"` + ProviderID string `json:"provider_id"` // "openai", "anthropic", etc. + Model string `json:"model"` + Endpoint string `json:"-"` // internal, not exposed in API + Status models.ProviderStatus `json:"status"` + Reason string `json:"reason"` // why this candidate was selected +} + +// ── Routing Decision ─────────────────────── + +// Decision records which candidate was selected and why. +// Stored in the usage_log for observability. +type Decision struct { + PolicyID string `json:"policy_id,omitempty"` // which policy matched + PolicyName string `json:"policy_name,omitempty"` // human-readable + PolicyType string `json:"policy_type,omitempty"` + Selected string `json:"selected"` // config ID that served the request + FallbackDepth int `json:"fallback_depth"` // 0 = first choice, 1+ = fallback + Candidates int `json:"candidates"` // total candidates evaluated +} diff --git a/server/store/interfaces.go b/server/store/interfaces.go index d1ee413..264b0a1 100644 --- a/server/store/interfaces.go +++ b/server/store/interfaces.go @@ -45,6 +45,7 @@ type Stores struct { Workspaces WorkspaceStore GitCredentials GitCredentialStore CapOverrides CapabilityOverrideStore + RoutingPolicies RoutingPolicyStore } // ========================================= @@ -587,6 +588,33 @@ type CapabilityOverrideStore interface { DeleteForProvider(ctx context.Context, providerConfigID string) error } +// ========================================= +// ROUTING POLICY STORE (v0.22.2) +// ========================================= + +type RoutingPolicyStore interface { + // Create adds a new routing policy. + Create(ctx context.Context, p *models.RoutingPolicy) error + + // Update modifies an existing routing policy. + Update(ctx context.Context, p *models.RoutingPolicy) error + + // Delete removes a routing policy by ID. + Delete(ctx context.Context, id string) error + + // GetByID returns a single policy. + GetByID(ctx context.Context, id string) (*models.RoutingPolicy, error) + + // ListActive returns all active policies, ordered by priority ASC. + ListActive(ctx context.Context) ([]models.RoutingPolicy, error) + + // ListAll returns all policies (including inactive), for admin listing. + ListAll(ctx context.Context) ([]models.RoutingPolicy, error) + + // ListForTeam returns active policies applicable to a team (team-scoped + global). + ListForTeam(ctx context.Context, teamID string) ([]models.RoutingPolicy, error) +} + // ========================================= // SHARED TYPES // ========================================= diff --git a/server/store/postgres/routing_policy.go b/server/store/postgres/routing_policy.go new file mode 100644 index 0000000..6573bc1 --- /dev/null +++ b/server/store/postgres/routing_policy.go @@ -0,0 +1,110 @@ +package postgres + +import ( + "context" + "database/sql" + "encoding/json" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +// RoutingPolicyStore implements store.RoutingPolicyStore for Postgres. +type RoutingPolicyStore struct { + db *sql.DB +} + +func NewRoutingPolicyStore(db *sql.DB) *RoutingPolicyStore { + return &RoutingPolicyStore{db: db} +} + +func (s *RoutingPolicyStore) Create(ctx context.Context, p *models.RoutingPolicy) error { + configJSON, err := json.Marshal(p.Config) + if err != nil { + return err + } + return s.db.QueryRowContext(ctx, ` + INSERT INTO routing_policies (name, scope, team_id, priority, policy_type, config, is_active) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING id, created_at, updated_at + `, p.Name, p.Scope, p.TeamID, p.Priority, p.Type, configJSON, p.IsActive, + ).Scan(&p.ID, &p.CreatedAt, &p.UpdatedAt) +} + +func (s *RoutingPolicyStore) Update(ctx context.Context, p *models.RoutingPolicy) error { + configJSON, err := json.Marshal(p.Config) + if err != nil { + return err + } + _, err = s.db.ExecContext(ctx, ` + UPDATE routing_policies + SET name = $2, scope = $3, team_id = $4, priority = $5, + policy_type = $6, config = $7, is_active = $8 + WHERE id = $1 + `, p.ID, p.Name, p.Scope, p.TeamID, p.Priority, p.Type, configJSON, p.IsActive) + return err +} + +func (s *RoutingPolicyStore) Delete(ctx context.Context, id string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM routing_policies WHERE id = $1`, id) + return err +} + +func (s *RoutingPolicyStore) GetByID(ctx context.Context, id string) (*models.RoutingPolicy, error) { + var p models.RoutingPolicy + var configJSON []byte + err := s.db.QueryRowContext(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies WHERE id = $1 + `, id).Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &p.IsActive, &p.CreatedAt, &p.UpdatedAt) + if err != nil { + return nil, err + } + json.Unmarshal(configJSON, &p.Config) + return &p, nil +} + +func (s *RoutingPolicyStore) ListActive(ctx context.Context) ([]models.RoutingPolicy, error) { + return s.query(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies + WHERE is_active = true + ORDER BY priority ASC, created_at ASC + `) +} + +func (s *RoutingPolicyStore) ListAll(ctx context.Context) ([]models.RoutingPolicy, error) { + return s.query(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies + ORDER BY priority ASC, created_at ASC + `) +} + +func (s *RoutingPolicyStore) ListForTeam(ctx context.Context, teamID string) ([]models.RoutingPolicy, error) { + return s.query(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies + WHERE is_active = true AND (scope = 'global' OR (scope = 'team' AND team_id = $1)) + ORDER BY priority ASC, created_at ASC + `, teamID) +} + +func (s *RoutingPolicyStore) query(ctx context.Context, q string, args ...interface{}) ([]models.RoutingPolicy, error) { + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var result []models.RoutingPolicy + for rows.Next() { + var p models.RoutingPolicy + var configJSON []byte + if err := rows.Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &p.IsActive, &p.CreatedAt, &p.UpdatedAt); err != nil { + return nil, err + } + json.Unmarshal(configJSON, &p.Config) + result = append(result, p) + } + return result, rows.Err() +} diff --git a/server/store/postgres/stores.go b/server/store/postgres/stores.go index d571bf7..d020e3e 100644 --- a/server/store/postgres/stores.go +++ b/server/store/postgres/stores.go @@ -38,5 +38,6 @@ func NewStores(db *sql.DB) store.Stores { Workspaces: NewWorkspaceStore(), GitCredentials: &GitCredentialStore{}, CapOverrides: NewCapOverrideStore(db), + RoutingPolicies: NewRoutingPolicyStore(db), } } diff --git a/server/store/sqlite/routing_policy.go b/server/store/sqlite/routing_policy.go new file mode 100644 index 0000000..a208526 --- /dev/null +++ b/server/store/sqlite/routing_policy.go @@ -0,0 +1,116 @@ +package sqlite + +import ( + "context" + "encoding/json" + "time" + + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/store" +) + +// RoutingPolicyStore implements store.RoutingPolicyStore for SQLite. +type RoutingPolicyStore struct{} + +func NewRoutingPolicyStore() *RoutingPolicyStore { return &RoutingPolicyStore{} } + +func (s *RoutingPolicyStore) Create(ctx context.Context, p *models.RoutingPolicy) error { + p.ID = store.NewID() + configJSON, err := json.Marshal(p.Config) + if err != nil { + return err + } + now := time.Now().UTC() + p.CreatedAt = now + p.UpdatedAt = now + _, err = DB.ExecContext(ctx, ` + INSERT INTO routing_policies (id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, p.ID, p.Name, p.Scope, p.TeamID, p.Priority, p.Type, string(configJSON), boolToInt(p.IsActive), p.CreatedAt.Format(time.RFC3339), p.UpdatedAt.Format(time.RFC3339)) + return err +} + +func (s *RoutingPolicyStore) Update(ctx context.Context, p *models.RoutingPolicy) error { + configJSON, err := json.Marshal(p.Config) + if err != nil { + return err + } + _, err = DB.ExecContext(ctx, ` + UPDATE routing_policies + SET name = ?, scope = ?, team_id = ?, priority = ?, + policy_type = ?, config = ?, is_active = ?, updated_at = datetime('now') + WHERE id = ? + `, p.Name, p.Scope, p.TeamID, p.Priority, p.Type, string(configJSON), boolToInt(p.IsActive), p.ID) + return err +} + +func (s *RoutingPolicyStore) Delete(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `DELETE FROM routing_policies WHERE id = ?`, id) + return err +} + +func (s *RoutingPolicyStore) GetByID(ctx context.Context, id string) (*models.RoutingPolicy, error) { + var p models.RoutingPolicy + var configJSON string + var isActive int + err := DB.QueryRowContext(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies WHERE id = ? + `, id).Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &isActive, &p.CreatedAt, &p.UpdatedAt) + if err != nil { + return nil, err + } + p.IsActive = isActive != 0 + json.Unmarshal([]byte(configJSON), &p.Config) + return &p, nil +} + +func (s *RoutingPolicyStore) ListActive(ctx context.Context) ([]models.RoutingPolicy, error) { + return s.query(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies + WHERE is_active = 1 + ORDER BY priority ASC, created_at ASC + `) +} + +func (s *RoutingPolicyStore) ListAll(ctx context.Context) ([]models.RoutingPolicy, error) { + return s.query(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies + ORDER BY priority ASC, created_at ASC + `) +} + +func (s *RoutingPolicyStore) ListForTeam(ctx context.Context, teamID string) ([]models.RoutingPolicy, error) { + return s.query(ctx, ` + SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at + FROM routing_policies + WHERE is_active = 1 AND (scope = 'global' OR (scope = 'team' AND team_id = ?)) + ORDER BY priority ASC, created_at ASC + `, teamID) +} + +func (s *RoutingPolicyStore) query(ctx context.Context, q string, args ...interface{}) ([]models.RoutingPolicy, error) { + rows, err := DB.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var result []models.RoutingPolicy + for rows.Next() { + var p models.RoutingPolicy + var configJSON string + var isActive int + if err := rows.Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &isActive, &p.CreatedAt, &p.UpdatedAt); err != nil { + return nil, err + } + p.IsActive = isActive != 0 + json.Unmarshal([]byte(configJSON), &p.Config) + result = append(result, p) + } + return result, rows.Err() +} + +var _ store.RoutingPolicyStore = (*RoutingPolicyStore)(nil) diff --git a/server/store/sqlite/stores.go b/server/store/sqlite/stores.go index 5c5dcfe..ba89120 100644 --- a/server/store/sqlite/stores.go +++ b/server/store/sqlite/stores.go @@ -38,5 +38,6 @@ func NewStores(db *sql.DB) store.Stores { Workspaces: NewWorkspaceStore(), GitCredentials: &GitCredentialStore{}, CapOverrides: NewCapOverrideStore(), + RoutingPolicies: NewRoutingPolicyStore(), } }