Changeset 0.22.2 (#96)

This commit is contained in:
2026-03-02 10:31:18 +00:00
parent cae6fd9f93
commit a44c768741
20 changed files with 1563 additions and 22 deletions

View File

@@ -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;

View File

@@ -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;

View File

@@ -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 {

View File

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

View File

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

View File

@@ -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"`
}

40
server/routing/convert.go Normal file
View File

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

280
server/routing/evaluator.go Normal file
View File

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

View File

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

View File

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

107
server/routing/types.go Normal file
View File

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

View File

@@ -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
// =========================================

View File

@@ -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()
}

View File

@@ -38,5 +38,6 @@ func NewStores(db *sql.DB) store.Stores {
Workspaces: NewWorkspaceStore(),
GitCredentials: &GitCredentialStore{},
CapOverrides: NewCapOverrideStore(db),
RoutingPolicies: NewRoutingPolicyStore(db),
}
}

View File

@@ -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)

View File

@@ -38,5 +38,6 @@ func NewStores(db *sql.DB) store.Stores {
Workspaces: NewWorkspaceStore(),
GitCredentials: &GitCredentialStore{},
CapOverrides: NewCapOverrideStore(),
RoutingPolicies: NewRoutingPolicyStore(),
}
}