fix: proxy logging, Anthropic translation, and admin API validation

Record upstream and gateway errors in request logs, including stream
interruptions. Translate Anthropic streamed tool calls, count cache
creation tokens, and drop top_p when temperature is set. Coalesce the
stats summary so an empty window no longer returns 500. Validate
provider and setting writes, allow clearing model overrides, add
provider update endpoint, and purge logs at startup.
This commit is contained in:
2026-10-03 16:17:11 +02:00
parent dca51c52e5
commit 5988e01906
19 changed files with 1100 additions and 172 deletions

View File

@@ -42,7 +42,8 @@ func run() error {
return fmt.Errorf("load config: %w", err)
}
slog.SetDefault(newLogger(cfg))
level := new(slog.LevelVar)
slog.SetDefault(newLogger(cfg, level))
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
@@ -67,6 +68,12 @@ func run() error {
st := store.New(pool)
mgr := providermgr.New(st, cipher)
// The dashboard's log level setting overrides the env default once set.
var levelName string
if ok, _ := st.GetSetting(ctx, "app.log_level", &levelName); ok {
level.Set(parseLogLevel(levelName))
}
if err := mgr.Reload(ctx); err != nil {
slog.Warn("initial provider load failed", "err", err)
}
@@ -104,9 +111,30 @@ func run() error {
go syncer.Run(ctx, cfg.Catalog.SyncInterval, refresh)
// Purge request logs on a daily tick. Retention is controlled by the
// log.retention_days setting (0 = keep forever). Defaults to 365 days.
// Purge request logs once at startup and then daily. Retention is
// controlled by the log.retention_days setting (0 = keep forever).
go func() {
purge := func() {
purgeCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
days := 365
if ok, _ := st.GetSetting(purgeCtx, "log.retention_days", &days); !ok {
days = 365
}
if days <= 0 {
return
}
tag, err := pool.Exec(purgeCtx,
`DELETE FROM request_logs WHERE created_at < NOW() - make_interval(days => $1)`, days)
if err != nil {
slog.Warn("request log purge failed", "err", err)
return
}
if tag.RowsAffected() > 0 {
slog.Info("purged request logs", "rows", tag.RowsAffected(), "retention_days", days)
}
}
purge()
tick := time.NewTicker(24 * time.Hour)
defer tick.Stop()
for {
@@ -114,17 +142,7 @@ func run() error {
case <-ctx.Done():
return
case <-tick.C:
var days int
purgeCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
if ok, _ := st.GetSetting(purgeCtx, "log.retention_days", &days); !ok {
days = 365
}
if days > 0 {
_, _ = pool.Exec(purgeCtx,
`DELETE FROM request_logs WHERE created_at < NOW() - ($1 || ' days')::interval`,
days)
}
cancel()
purge()
}
}
}()
@@ -142,6 +160,7 @@ func run() error {
PublicBaseURL: cfg.Server.PublicURL,
Limiter: limiter,
SyncNow: syncNow,
LogLevel: level,
}
a := api.New(deps)
@@ -178,15 +197,29 @@ func run() error {
return nil
}
func newLogger(cfg *config.Config) *slog.Logger {
opts := &slog.HandlerOptions{Level: slog.LevelInfo}
func newLogger(cfg *config.Config, level *slog.LevelVar) *slog.Logger {
level.Set(slog.LevelInfo)
opts := &slog.HandlerOptions{Level: level}
if cfg.IsDev() {
opts.Level = slog.LevelDebug
level.Set(slog.LevelDebug)
return slog.New(slog.NewTextHandler(os.Stdout, opts))
}
return slog.New(slog.NewJSONHandler(os.Stdout, opts))
}
func parseLogLevel(s string) slog.Level {
switch s {
case "debug":
return slog.LevelDebug
case "warn":
return slog.LevelWarn
case "error":
return slog.LevelError
default:
return slog.LevelInfo
}
}
func newRouter(pool interface{ Ping(context.Context) error }, a *api.API) http.Handler {
r := chi.NewRouter()
r.Use(middleware.RequestID)

View File

@@ -5,13 +5,19 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log/slog"
"net/http"
"slices"
"strconv"
"strings"
"time"
"github.com/go-chi/chi/v5"
"github.com/google/uuid"
"git.govekar.net/LovroG05/localmodelroutex/internal/catalog"
"git.govekar.net/LovroG05/localmodelroutex/internal/providers"
"git.govekar.net/LovroG05/localmodelroutex/internal/store"
)
@@ -23,16 +29,73 @@ func (a *API) adminListProviders(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
counts, err := a.Store.ProviderModelCounts(r.Context())
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
out := make([]map[string]any, 0, len(rows))
for _, p := range rows {
var cfg map[string]any
if len(p.Config) > 0 {
_ = json.Unmarshal(p.Config, &cfg)
}
if cfg == nil {
cfg = map[string]any{}
}
c := counts[p.ID]
// A provider that failed to build (bad credentials) is enabled in the DB
// but absent from the live manager.
_, live := a.Manager.Get(p.Name)
out = append(out, map[string]any{
"id": p.ID, "name": p.Name, "type": p.Type, "base_url": p.BaseURL,
"enabled": p.Enabled, "has_credentials": len(p.Credentials) > 0,
"config": cfg, "created_at": p.CreatedAt,
"live": live,
"model_count": c.Models,
"available_count": c.Available,
"last_synced_at": c.LastSyncedAt,
})
}
writeJSON(w, http.StatusOK, out)
}
// validateProviderName rejects names that cannot serve as an address prefix.
func validateProviderName(name string) error {
if name == "" {
return fmt.Errorf("name is required")
}
if len(name) > 64 {
return fmt.Errorf("name must be 64 characters or fewer")
}
if strings.EqualFold(name, catalog.AnyProvider) {
return fmt.Errorf("%q is reserved for cross-provider routing", name)
}
for _, c := range name {
ok := (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.'
if !ok {
return fmt.Errorf("name may only contain letters, digits, '-', '_' and '.'")
}
}
return nil
}
// sealCredentials encrypts a non-empty credential map. Empty maps return nil.
func (a *API) sealCredentials(creds map[string]any) ([]byte, error) {
clean := map[string]any{}
for k, v := range creds {
if s, ok := v.(string); ok && strings.TrimSpace(s) == "" {
continue
}
clean[k] = v
}
if len(clean) == 0 {
return nil, nil
}
plain, _ := json.Marshal(clean)
return a.Cipher.Seal(plain)
}
func (a *API) adminCreateProvider(w http.ResponseWriter, r *http.Request) {
var in struct {
Name string `json:"name"`
@@ -46,20 +109,25 @@ func (a *API) adminCreateProvider(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusBadRequest, "invalid JSON")
return
}
if in.Name == "" || in.Type == "" {
writeError(w, http.StatusBadRequest, "name and type are required")
in.Name = strings.TrimSpace(in.Name)
in.BaseURL = strings.TrimSpace(in.BaseURL)
if err := validateProviderName(in.Name); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if !slices.Contains(providers.Types(), in.Type) {
writeError(w, http.StatusBadRequest, fmt.Sprintf("unknown provider type %q (want one of %s)", in.Type, strings.Join(providers.Types(), ", ")))
return
}
if in.BaseURL != "" && !strings.HasPrefix(in.BaseURL, "http://") && !strings.HasPrefix(in.BaseURL, "https://") {
writeError(w, http.StatusBadRequest, "base_url must start with http:// or https://")
return
}
var creds []byte
if len(in.Credentials) > 0 {
plain, _ := json.Marshal(in.Credentials)
sealed, err := a.Cipher.Seal(plain)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to encrypt credentials")
return
}
creds = sealed
creds, err := a.sealCredentials(in.Credentials)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to encrypt credentials")
return
}
var cfg []byte
if len(in.Config) > 0 {
@@ -70,11 +138,26 @@ func (a *API) adminCreateProvider(w http.ResponseWriter, r *http.Request) {
enabled = *in.Enabled
}
// Build once up front so a bad credential shape is reported to the caller
// instead of being logged on the next reload.
var plain []byte
if creds != nil {
plain, _ = json.Marshal(in.Credentials)
}
if _, err := providers.New(in.Type, providers.Config{Name: in.Name, BaseURL: in.BaseURL, Credentials: plain, Settings: in.Config}); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
id, err := a.Store.CreateProvider(r.Context(), store.CreateProviderParams{
Name: in.Name, Type: in.Type, BaseURL: in.BaseURL,
Credentials: creds, Enabled: enabled, Config: cfg,
})
if err != nil {
if strings.Contains(err.Error(), "providers_name_key") {
writeError(w, http.StatusConflict, fmt.Sprintf("a provider named %q already exists", in.Name))
return
}
writeError(w, http.StatusInternalServerError, err.Error())
return
}
@@ -82,6 +165,50 @@ func (a *API) adminCreateProvider(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusCreated, map[string]any{"id": id})
}
func (a *API) adminUpdateProvider(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(chi.URLParam(r, "id"))
if err != nil {
writeError(w, http.StatusBadRequest, "invalid id")
return
}
var in struct {
BaseURL *string `json:"base_url"`
Enabled *bool `json:"enabled"`
Credentials map[string]any `json:"credentials"`
Config map[string]any `json:"config"`
}
if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON")
return
}
p := store.UpdateProviderParams{BaseURL: in.BaseURL, Enabled: in.Enabled}
if in.BaseURL != nil {
u := strings.TrimSpace(*in.BaseURL)
if u != "" && !strings.HasPrefix(u, "http://") && !strings.HasPrefix(u, "https://") {
writeError(w, http.StatusBadRequest, "base_url must start with http:// or https://")
return
}
p.BaseURL = &u
}
if len(in.Credentials) > 0 {
creds, err := a.sealCredentials(in.Credentials)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to encrypt credentials")
return
}
p.Credentials = creds
}
if in.Config != nil {
p.Config, _ = json.Marshal(in.Config)
}
if err := a.Store.UpdateProvider(r.Context(), id, p); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
a.syncNow(r)
w.WriteHeader(http.StatusNoContent)
}
func (a *API) adminDeleteProvider(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(chi.URLParam(r, "id"))
if err != nil {
@@ -109,6 +236,31 @@ func (a *API) adminCreateKey(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusBadRequest, "invalid JSON")
return
}
in.Name = strings.TrimSpace(in.Name)
if in.Name == "" {
writeError(w, http.StatusBadRequest, "name is required")
return
}
if in.ExpiresInDays < 0 {
writeError(w, http.StatusBadRequest, "expires_in_days must be 0 or more")
return
}
rules := make([]string, 0, len(in.Rules))
for _, rule := range in.Rules {
rule = strings.TrimSpace(rule)
if rule == "" {
continue
}
if rule != "*" && !strings.Contains(rule, "/") {
writeError(w, http.StatusBadRequest, fmt.Sprintf("rule %q must be '*', '<provider>/*' or '<provider>/<slug>'", rule))
return
}
rules = append(rules, rule)
}
if !in.AllowAllModels && len(rules) == 0 {
writeError(w, http.StatusBadRequest, "a restricted key needs at least one model rule")
return
}
raw, err := generateKey()
if err != nil {
@@ -133,7 +285,7 @@ func (a *API) adminCreateKey(w http.ResponseWriter, r *http.Request) {
id, err := a.Store.CreateVirtualKey(r.Context(), store.CreateVirtualKeyParams{
Name: in.Name, KeyHash: sum[:], KeyPrefix: prefix, UserID: userID,
AllowAllModels: in.AllowAllModels, ExpiresAt: expires, Rules: in.Rules,
AllowAllModels: in.AllowAllModels, ExpiresAt: expires, Rules: rules,
})
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
@@ -144,22 +296,21 @@ func (a *API) adminCreateKey(w http.ResponseWriter, r *http.Request) {
}
func (a *API) adminListKeys(w http.ResponseWriter, r *http.Request) {
// Admins see all keys; regular users see only their own.
au := userFrom(r.Context())
var userID *uuid.UUID
if au != nil && !au.IsAdmin {
userID = &au.User.ID
}
keys, err := a.Store.ListVirtualKeys(r.Context(), userID)
keys, err := a.Store.ListVirtualKeys(r.Context(), scopedUserID(r))
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
out := make([]map[string]any, 0, len(keys))
for _, k := range keys {
rules := k.Rules
if rules == nil {
rules = []string{}
}
out = append(out, map[string]any{
"id": k.ID, "name": k.Name, "prefix": k.KeyPrefix, "type": k.Type,
"allow_all_models": k.AllowAllModels, "expires_at": k.ExpiresAt,
"allow_all_models": k.AllowAllModels, "rules": rules,
"created_at": k.CreatedAt, "expires_at": k.ExpiresAt,
"revoked_at": k.RevokedAt, "last_used_at": k.LastUsedAt,
})
}
@@ -202,38 +353,95 @@ func (a *API) adminListModels(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if rows == nil {
rows = []store.AdminModelRow{}
}
writeJSON(w, http.StatusOK, rows)
}
// patchField decodes an optional JSON field. A present null clears the column.
func patchField[T any](fields map[string]json.RawMessage, key string) (store.Patch[T], error) {
raw, ok := fields[key]
if !ok {
return store.Patch[T]{}, nil
}
if string(raw) == "null" {
return store.Patch[T]{Set: true}, nil
}
var v T
if err := json.Unmarshal(raw, &v); err != nil {
return store.Patch[T]{}, fmt.Errorf("invalid value for %s", key)
}
return store.Patch[T]{Set: true, Value: &v}, nil
}
func (a *API) adminUpdateModel(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(chi.URLParam(r, "id"))
if err != nil {
writeError(w, http.StatusBadRequest, "invalid id")
return
}
var in struct {
Enabled *bool `json:"enabled"`
Blocked *bool `json:"blocked"`
ParamOverrides map[string]any `json:"param_overrides"`
InputPrice *float64 `json:"input_price_per_mtok"`
OutputPrice *float64 `json:"output_price_per_mtok"`
CachedPrice *float64 `json:"cached_price_per_mtok"`
RPMLimit *int `json:"rpm_limit"`
TPMLimit *int `json:"tpm_limit"`
}
if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
var fields map[string]json.RawMessage
if err := json.NewDecoder(r.Body).Decode(&fields); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON")
return
}
var params []byte
if in.ParamOverrides != nil {
params, _ = json.Marshal(in.ParamOverrides)
var o store.ModelOverride
if raw, ok := fields["enabled"]; ok {
if err := json.Unmarshal(raw, &o.Enabled); err != nil {
writeError(w, http.StatusBadRequest, "invalid value for enabled")
return
}
}
if err := a.Store.UpdateModelOverride(r.Context(), id, store.ModelOverride{
Enabled: in.Enabled, Blocked: in.Blocked, ParamOverrides: params,
InputPrice: in.InputPrice, OutputPrice: in.OutputPrice, CachedPrice: in.CachedPrice,
RPMLimit: in.RPMLimit, TPMLimit: in.TPMLimit,
}); err != nil {
if raw, ok := fields["blocked"]; ok {
if err := json.Unmarshal(raw, &o.Blocked); err != nil {
writeError(w, http.StatusBadRequest, "invalid value for blocked")
return
}
}
if raw, ok := fields["param_overrides"]; ok && string(raw) != "null" {
var m map[string]any
if err := json.Unmarshal(raw, &m); err != nil {
writeError(w, http.StatusBadRequest, "param_overrides must be an object")
return
}
o.ParamOverrides, _ = json.Marshal(m)
}
if o.InputPrice, err = patchField[float64](fields, "input_price_per_mtok"); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if o.OutputPrice, err = patchField[float64](fields, "output_price_per_mtok"); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if o.CachedPrice, err = patchField[float64](fields, "cached_price_per_mtok"); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if o.RPMLimit, err = patchField[int](fields, "rpm_limit"); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if o.TPMLimit, err = patchField[int](fields, "tpm_limit"); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
for _, p := range []store.Patch[float64]{o.InputPrice, o.OutputPrice, o.CachedPrice} {
if p.Value != nil && *p.Value < 0 {
writeError(w, http.StatusBadRequest, "prices must be 0 or more")
return
}
}
for _, p := range []store.Patch[int]{o.RPMLimit, o.TPMLimit} {
if p.Value != nil && *p.Value < 0 {
writeError(w, http.StatusBadRequest, "limits must be 0 or more")
return
}
}
if err := a.Store.UpdateModelOverride(r.Context(), id, o); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
@@ -243,42 +451,92 @@ func (a *API) adminUpdateModel(w http.ResponseWriter, r *http.Request) {
// --- settings ---
// Settings the dashboard may read and write, with their validators.
var settingValidators = map[string]func(any) error{
"routing.strategy": oneOf(string(catalog.StrategyRoundRobin), string(catalog.StrategyLowestCost)),
"app.name": func(v any) error {
s, ok := v.(string)
if !ok || strings.TrimSpace(s) == "" || len(s) > 64 {
return fmt.Errorf("app.name must be a non-empty string of at most 64 characters")
}
return nil
},
"app.log_level": oneOf("debug", "info", "warn", "error"),
"log.retention_days": func(v any) error {
f, ok := v.(float64)
if !ok || f < 0 || f != float64(int(f)) {
return fmt.Errorf("log.retention_days must be a whole number of 0 or more")
}
return nil
},
}
func oneOf(allowed ...string) func(any) error {
return func(v any) error {
s, _ := v.(string)
if !slices.Contains(allowed, s) {
return fmt.Errorf("value must be one of %s", strings.Join(allowed, ", "))
}
return nil
}
}
func (a *API) adminGetSettings(w http.ResponseWriter, r *http.Request) {
out := map[string]any{}
var strategy string
if ok, _ := a.Store.GetSetting(r.Context(), "routing.strategy", &strategy); ok {
out["routing.strategy"] = strategy
}
var appName string
if ok, _ := a.Store.GetSetting(r.Context(), "app.name", &appName); ok {
out["app.name"] = appName
}
var logLevel string
if ok, _ := a.Store.GetSetting(r.Context(), "app.log_level", &logLevel); ok {
out["app.log_level"] = logLevel
}
var retentionDays int
if ok, _ := a.Store.GetSetting(r.Context(), "log.retention_days", &retentionDays); ok {
out["log.retention_days"] = retentionDays
for key := range settingValidators {
var v any
if ok, _ := a.Store.GetSetting(r.Context(), key, &v); ok {
out[key] = v
}
}
writeJSON(w, http.StatusOK, out)
}
func (a *API) adminPutSetting(w http.ResponseWriter, r *http.Request) {
key := chi.URLParam(r, "key")
validate, ok := settingValidators[key]
if !ok {
writeError(w, http.StatusNotFound, fmt.Sprintf("unknown setting %q", key))
return
}
var value any
if err := json.NewDecoder(r.Body).Decode(&value); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON")
return
}
if err := validate(value); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if err := a.Store.SetSetting(r.Context(), key, value); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
_ = a.Cache.Reload(r.Context())
switch key {
case "routing.strategy":
_ = a.Cache.Reload(r.Context())
case "app.log_level":
if a.LogLevel != nil {
a.LogLevel.Set(parseLogLevel(value.(string)))
}
}
w.WriteHeader(http.StatusNoContent)
}
// parseLogLevel maps a setting value to a slog level, defaulting to info.
func parseLogLevel(s string) slog.Level {
switch s {
case "debug":
return slog.LevelDebug
case "warn":
return slog.LevelWarn
case "error":
return slog.LevelError
default:
return slog.LevelInfo
}
}
// --- request logs ---
func (a *API) adminListLogs(w http.ResponseWriter, r *http.Request) {
@@ -332,12 +590,7 @@ func (a *API) adminGetLogContent(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusBadRequest, "invalid log id")
return
}
au := userFrom(r.Context())
var userID *uuid.UUID
if au != nil && !au.IsAdmin {
userID = &au.User.ID
}
messages, completion, err := a.Store.GetLogContent(r.Context(), id, userID)
messages, completion, err := a.Store.GetLogContent(r.Context(), id, scopedUserID(r))
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
@@ -390,7 +643,7 @@ func (a *API) adminStatsSummary(w http.ResponseWriter, r *http.Request) {
func (a *API) adminStatsModels(w http.ResponseWriter, r *http.Request) {
limit := 6
if l := r.URL.Query().Get("limit"); l != "" {
if n, err := strconv.Atoi(l); err == nil && n > 0 {
if n, err := strconv.Atoi(l); err == nil && n > 0 && n <= 50 {
limit = n
}
}
@@ -408,7 +661,7 @@ func (a *API) adminStatsModels(w http.ResponseWriter, r *http.Request) {
func (a *API) adminStatsSeries(w http.ResponseWriter, r *http.Request) {
buckets := 24
if b := r.URL.Query().Get("buckets"); b != "" {
if n, err := strconv.Atoi(b); err == nil && n > 0 {
if n, err := strconv.Atoi(b); err == nil && n > 0 && n <= 366 {
buckets = n
}
}
@@ -437,7 +690,9 @@ func (a *API) adminRefreshCatalog(w http.ResponseWriter, r *http.Request) {
// syncNow triggers an immediate provider + catalog refresh if wired.
func (a *API) syncNow(r *http.Request) {
if a.SyncNow != nil {
_ = a.SyncNow(r.Context())
if err := a.SyncNow(r.Context()); err != nil {
slog.Warn("catalog sync after admin change failed", "err", err)
}
}
}

View File

@@ -4,6 +4,7 @@ package api
import (
"context"
"log/slog"
"net/http"
"github.com/go-chi/chi/v5"
@@ -44,10 +45,10 @@ func userFrom(ctx context.Context) *authedUser {
// Deps are the dependencies the API handlers need.
type Deps struct {
Store *store.Store
Cache *catalog.Cache
Manager *providermgr.Manager
Cipher *crypto.Cipher
Store *store.Store
Cache *catalog.Cache
Manager *providermgr.Manager
Cipher *crypto.Cipher
OIDCVerifier oidcVerifier
OIDCClientID string
OIDCClientSecret string
@@ -58,10 +59,13 @@ type Deps struct {
// build the redirect_uri and the post-login fragment redirect so neither
// can be overridden by a forged Host/X-Forwarded-Host header.
PublicBaseURL string
Limiter *ratelimit.Limiter
Limiter *ratelimit.Limiter
// SyncNow triggers an immediate provider reload + catalog sync (used after
// admin changes). May be nil.
SyncNow func(context.Context) error
// LogLevel is the live logger level, updated when the app.log_level
// setting changes. May be nil.
LogLevel *slog.LevelVar
}
// API holds handler dependencies.
@@ -110,6 +114,7 @@ func (a *API) Mount(r chi.Router) {
r.Use(a.requireAdmin)
r.Get("/providers", a.adminListProviders)
r.Post("/providers", a.adminCreateProvider)
r.Patch("/providers/{id}", a.adminUpdateProvider)
r.Delete("/providers/{id}", a.adminDeleteProvider)
r.Get("/models", a.adminListModels)
r.Patch("/models/{id}", a.adminUpdateModel)
@@ -124,13 +129,20 @@ func (a *API) Mount(r chi.Router) {
r.Get("/api/auth/config", a.authConfig)
}
// apiMe returns the current user's profile and admin status.
// apiMe returns the current user's profile, admin status, and the configured
// app name so the dashboard can brand itself without admin access.
func (a *API) apiMe(w http.ResponseWriter, r *http.Request) {
u := userFrom(r.Context())
appName := "LMR"
var s string
if ok, _ := a.Store.GetSetting(r.Context(), "app.name", &s); ok && s != "" {
appName = s
}
writeJSON(w, http.StatusOK, map[string]any{
"id": u.User.ID,
"email": u.User.Email,
"name": u.User.Name,
"is_admin": u.IsAdmin,
"app_name": appName,
})
}

View File

@@ -87,16 +87,23 @@ func (a *API) chatCompletions(w http.ResponseWriter, r *http.Request) {
resp, err := cc.ChatCompletion(r.Context(), req)
if err != nil {
writeError(w, http.StatusBadGateway, err.Error())
a.logRequest(r, key, entry, address, http.StatusBadGateway, false, providers.Usage{}, err.Error(), start, nil, messages, "")
return
}
a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens)
respBody := injectCreated(resp.Body)
respBody := resp.Body
errMsg := ""
if resp.StatusCode >= 400 {
errMsg = upstreamErrorMessage(resp.Body)
} else {
respBody = injectCreated(resp.Body)
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(resp.StatusCode)
_, _ = w.Write(respBody)
completion := extractCompletion(resp.Body)
a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, "", start, nil, messages, completion)
a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, errMsg, start, nil, messages, completion)
}
func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.ChatCompleter, req *providers.ChatRequest, entry *catalog.Entry, key *store.VirtualKey, address string, messages json.RawMessage, start time.Time) {
@@ -107,10 +114,11 @@ func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.Ch
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(apiErr.StatusCode)
_, _ = w.Write(apiErr.Body)
a.logRequest(r, key, entry, address, apiErr.StatusCode, true, providers.Usage{}, "upstream error", start, nil, messages, "")
a.logRequest(r, key, entry, address, apiErr.StatusCode, true, providers.Usage{}, upstreamErrorMessage(apiErr.Body), start, nil, messages, "")
return
}
writeError(w, http.StatusBadGateway, err.Error())
a.logRequest(r, key, entry, address, http.StatusBadGateway, true, providers.Usage{}, err.Error(), start, nil, messages, "")
return
}
defer stream.Close()
@@ -124,12 +132,20 @@ func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.Ch
var ttft *int
var reasoning, content strings.Builder
created := time.Now().Unix()
errMsg := ""
for {
ev, err := stream.Recv()
if errors.Is(err, io.EOF) {
break
}
if err != nil {
// The status line is already sent, so the error can only be
// recorded in the log row and signalled by closing the stream.
if r.Context().Err() != nil {
errMsg = "client disconnected"
} else {
errMsg = "stream interrupted: " + err.Error()
}
break
}
if ttft == nil {
@@ -157,7 +173,7 @@ func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.Ch
}
usage := stream.Usage()
a.Limiter.RecordTokens(entry.Address, usage.TotalTokens)
a.logRequest(r, key, entry, address, http.StatusOK, true, usage, "", start, ttft, messages, completion)
a.logRequest(r, key, entry, address, http.StatusOK, true, usage, errMsg, start, ttft, messages, completion)
}
func applyOverrides(payload, overrides map[string]any) {
@@ -205,24 +221,89 @@ func (a *API) logRequest(r *http.Request, key *store.VirtualKey, entry *catalog.
if err := a.Store.InsertLogContent(ctx, id, messages, completion); err != nil {
slog.Warn("failed to write request log content", "err", err)
}
if key != nil {
if key != nil && key.ID != uuid.Nil {
_ = a.Store.TouchVirtualKey(ctx, key.ID)
}
}()
}
// marshalMessages extracts the messages array from the request payload as raw JSON.
// marshalMessages extracts the messages array from the request payload as raw
// JSON for the content log. Inline image data is replaced by a short marker so
// a single multimodal request cannot write megabytes of base64 into the log.
func marshalMessages(v any) json.RawMessage {
if v == nil {
arr, ok := v.([]any)
if !ok {
return json.RawMessage("[]")
}
b, err := json.Marshal(v)
b, err := json.Marshal(redactImages(arr))
if err != nil {
return json.RawMessage("[]")
}
return b
}
// redactImages returns a copy of the messages with data: image URLs replaced.
func redactImages(messages []any) []any {
out := make([]any, len(messages))
for i, raw := range messages {
m, ok := raw.(map[string]any)
if !ok {
out[i] = raw
continue
}
parts, ok := m["content"].([]any)
if !ok {
out[i] = m
continue
}
cp := make(map[string]any, len(m))
maps.Copy(cp, m)
newParts := make([]any, len(parts))
for j, p := range parts {
part, ok := p.(map[string]any)
if !ok || part["type"] != "image_url" {
newParts[j] = p
continue
}
iu, _ := part["image_url"].(map[string]any)
url, _ := iu["url"].(string)
if !strings.HasPrefix(url, "data:") {
newParts[j] = p
continue
}
mediaType := url
if i := strings.IndexAny(url, ";,"); i > 0 {
mediaType = url[:i]
}
newParts[j] = map[string]any{
"type": "image_url",
"image_url": map[string]any{"url": fmt.Sprintf("[inline image %s, %d bytes omitted]", strings.TrimPrefix(mediaType, "data:"), len(url))},
}
}
cp["content"] = newParts
out[i] = cp
}
return out
}
// upstreamErrorMessage pulls the message out of an OpenAI-style error body,
// falling back to a truncated copy of the raw body.
func upstreamErrorMessage(body []byte) string {
var e struct {
Error struct {
Message string `json:"message"`
} `json:"error"`
}
if json.Unmarshal(body, &e) == nil && e.Error.Message != "" {
return e.Error.Message
}
msg := strings.TrimSpace(string(body))
if len(msg) > 500 {
msg = msg[:500] + "…"
}
return msg
}
// extractCompletion pulls the first choice's message content from a non-streaming
// response body. Reasoning (reasoning_content, reasoning, or reasoning_details) is
// wrapped in think tags and prepended when present.

View File

@@ -0,0 +1,48 @@
package api
import (
"strings"
"testing"
)
func TestRedactImagesReplacesDataURIs(t *testing.T) {
msgs := []any{
map[string]any{"role": "system", "content": "be brief"},
map[string]any{"role": "user", "content": []any{
map[string]any{"type": "text", "text": "what is this?"},
map[string]any{"type": "image_url", "image_url": map[string]any{"url": "data:image/png;base64," + strings.Repeat("A", 4000)}},
map[string]any{"type": "image_url", "image_url": map[string]any{"url": "https://example.com/a.png"}},
}},
}
out := string(marshalMessages(msgs))
if strings.Contains(out, strings.Repeat("A", 100)) {
t.Fatalf("base64 payload leaked into the log: %d bytes", len(out))
}
if !strings.Contains(out, "inline image image/png") {
t.Errorf("missing redaction marker: %s", out)
}
if !strings.Contains(out, "https://example.com/a.png") {
t.Errorf("remote URLs must be kept: %s", out)
}
if !strings.Contains(out, "be brief") || !strings.Contains(out, "what is this?") {
t.Errorf("text content must be kept: %s", out)
}
}
func TestMarshalMessagesNonArray(t *testing.T) {
if got := string(marshalMessages(nil)); got != "[]" {
t.Errorf("nil -> %q", got)
}
if got := string(marshalMessages("oops")); got != "[]" {
t.Errorf("string -> %q", got)
}
}
func TestUpstreamErrorMessage(t *testing.T) {
if got := upstreamErrorMessage([]byte(`{"error":{"message":"rate limited","type":"x"}}`)); got != "rate limited" {
t.Errorf("got %q", got)
}
if got := upstreamErrorMessage([]byte(` bad gateway `)); got != "bad gateway" {
t.Errorf("got %q", got)
}
}

View File

@@ -64,16 +64,21 @@ func (a *API) embeddings(w http.ResponseWriter, r *http.Request) {
resp, err := emb.Embeddings(r.Context(), req)
if err != nil {
writeError(w, http.StatusBadGateway, err.Error())
a.logRequest(r, key, entry, address, http.StatusBadGateway, false, providers.Usage{}, err.Error(), start, nil, nil, "")
return
}
a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens)
errMsg := ""
if resp.StatusCode >= 400 {
errMsg = upstreamErrorMessage(resp.Body)
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(resp.StatusCode)
_, _ = w.Write(injectCreated(resp.Body))
_, _ = w.Write(resp.Body)
a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, "", start, nil, nil, "")
a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, errMsg, start, nil, nil, "")
}
// ptrInt dereferences an *int, returning 0 (no limit) when nil.

View File

@@ -16,9 +16,9 @@ func newTestCache(strategy Strategy, entries ...*Entry) *Cache {
func TestSplitAddress(t *testing.T) {
cases := []struct {
in string
provider, slug string
ok bool
in string
provider, slug string
ok bool
}{
{"openai/gpt-4o", "openai", "gpt-4o", true},
{"openrouter/openai/gpt-4o", "openrouter", "openai/gpt-4o", true},

View File

@@ -21,9 +21,11 @@ type fakeProvider struct {
models []providers.Model
}
func (f *fakeProvider) Name() string { return f.name }
func (f *fakeProvider) Type() string { return f.typ }
func (f *fakeProvider) Capabilities() []providers.Capability { return []providers.Capability{providers.CapabilityChat} }
func (f *fakeProvider) Name() string { return f.name }
func (f *fakeProvider) Type() string { return f.typ }
func (f *fakeProvider) Capabilities() []providers.Capability {
return []providers.Capability{providers.CapabilityChat}
}
func (f *fakeProvider) ListModels(context.Context) ([]providers.Model, error) {
return f.models, nil
}

View File

@@ -89,15 +89,46 @@ func (sy *Syncer) Run(ctx context.Context, interval time.Duration, refresh func(
}
// normalize maps a provider listing entry to a DiscoveredModel, extracting
// pricing/context where the provider reports it (currently OpenRouter).
// display name, pricing and context length where the provider reports them.
func normalize(providerType string, m providers.Model) store.DiscoveredModel {
d := store.DiscoveredModel{Slug: m.ID, Raw: m.Raw}
if providerType == "openrouter" {
switch providerType {
case "openrouter":
extractOpenRouter(&d, m.Raw)
case "anthropic":
extractAnthropic(&d, m.Raw)
case "llamacpp":
extractLlamaCpp(&d, m.Raw)
}
return d
}
func extractAnthropic(d *store.DiscoveredModel, raw []byte) {
var entry struct {
DisplayName string `json:"display_name"`
}
if err := json.Unmarshal(raw, &entry); err != nil {
return
}
d.DisplayName = entry.DisplayName
}
// extractLlamaCpp reads the training context length that llama.cpp server
// reports under meta.n_ctx_train.
func extractLlamaCpp(d *store.DiscoveredModel, raw []byte) {
var entry struct {
Meta struct {
NCtxTrain *int `json:"n_ctx_train"`
} `json:"meta"`
}
if err := json.Unmarshal(raw, &entry); err != nil {
return
}
if entry.Meta.NCtxTrain != nil && *entry.Meta.NCtxTrain > 0 {
d.ContextLength = entry.Meta.NCtxTrain
}
}
func extractOpenRouter(d *store.DiscoveredModel, raw []byte) {
var entry struct {
Name string `json:"name"`

View File

@@ -22,15 +22,19 @@ type stream struct {
finished bool
finishReason string
usage providers.Usage
// toolIndex maps an Anthropic content block index to its OpenAI
// tool_calls index, assigned in order of appearance.
toolIndex map[int]int
}
func newStream(rc io.ReadCloser, model string) *stream {
return &stream{
rc: rc,
br: bufio.NewReaderSize(rc, 64*1024),
model: model,
id: "chatcmpl-stream",
created: time.Now().Unix(),
rc: rc,
br: bufio.NewReaderSize(rc, 64*1024),
model: model,
id: "chatcmpl-stream",
created: time.Now().Unix(),
toolIndex: map[int]int{},
}
}
@@ -71,8 +75,9 @@ func (s *stream) process(data []byte) (ev providers.Event, emit, done bool) {
Message struct {
ID string `json:"id"`
Usage struct {
InputTokens int `json:"input_tokens"`
CacheReadInputTokens int `json:"cache_read_input_tokens"`
InputTokens int `json:"input_tokens"`
CacheReadInputTokens int `json:"cache_read_input_tokens"`
CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
} `json:"usage"`
} `json:"message"`
}
@@ -80,15 +85,33 @@ func (s *stream) process(data []byte) (ev providers.Event, emit, done bool) {
if d.Message.ID != "" {
s.id = d.Message.ID
}
s.usage.PromptTokens = d.Message.Usage.InputTokens + d.Message.Usage.CacheReadInputTokens
s.usage.PromptTokens = d.Message.Usage.InputTokens + d.Message.Usage.CacheReadInputTokens + d.Message.Usage.CacheCreationInputTokens
s.usage.CachedTokens = d.Message.Usage.CacheReadInputTokens
case "content_block_start":
var d struct {
Index int `json:"index"`
ContentBlock struct {
Type string `json:"type"`
ID string `json:"id"`
Name string `json:"name"`
} `json:"content_block"`
}
_ = json.Unmarshal(data, &d)
if d.ContentBlock.Type == "tool_use" {
idx := len(s.toolIndex)
s.toolIndex[d.Index] = idx
return s.toolCallChunk(idx, d.ContentBlock.ID, d.ContentBlock.Name, ""), true, false
}
case "content_block_delta":
var d struct {
Index int `json:"index"`
Delta struct {
Type string `json:"type"`
Text string `json:"text"`
Thinking string `json:"thinking"`
Type string `json:"type"`
Text string `json:"text"`
Thinking string `json:"thinking"`
PartialJSON string `json:"partial_json"`
} `json:"delta"`
}
_ = json.Unmarshal(data, &d)
@@ -101,6 +124,10 @@ func (s *stream) process(data []byte) (ev providers.Event, emit, done bool) {
if d.Delta.Thinking != "" {
return s.reasoningChunk(d.Delta.Thinking), true, false
}
case "input_json_delta":
if idx, ok := s.toolIndex[d.Index]; ok && d.Delta.PartialJSON != "" {
return s.toolCallChunk(idx, "", "", d.Delta.PartialJSON), true, false
}
}
case "message_delta":
@@ -144,6 +171,25 @@ func (s *stream) reasoningChunk(text string) providers.Event {
return s.chunk(delta, nil, nil)
}
// toolCallChunk emits one OpenAI tool_calls delta. The first chunk for a call
// carries id and name; later chunks append to the JSON arguments string.
func (s *stream) toolCallChunk(index int, id, name, args string) providers.Event {
call := map[string]any{"index": index}
if id != "" {
call["id"] = id
call["type"] = "function"
call["function"] = map[string]any{"name": name, "arguments": ""}
} else {
call["function"] = map[string]any{"arguments": args}
}
delta := map[string]any{"tool_calls": []any{call}}
if !s.roleSent {
delta["role"] = "assistant"
s.roleSent = true
}
return s.chunk(delta, nil, nil)
}
func (s *stream) finalChunk() providers.Event {
s.usage.TotalTokens = s.usage.PromptTokens + s.usage.CompletionTokens
finish := s.finishReason

View File

@@ -41,8 +41,8 @@ func TestStreamTranslation(t *testing.T) {
Object string `json:"object"`
Choices []struct {
Delta struct {
Role string `json:"role"`
Content string `json:"content"`
Role string `json:"role"`
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
@@ -132,3 +132,76 @@ func TestStreamThinking(t *testing.T) {
t.Errorf("content = %q", content.String())
}
}
func TestStreamToolUse(t *testing.T) {
sse := strings.Join([]string{
`data: {"type":"message_start","message":{"id":"msg_2","usage":{"input_tokens":4,"cache_creation_input_tokens":6}}}`,
``,
`data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_1","name":"get_weather"}}`,
``,
`data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"city\":"}}`,
``,
`data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"\"Oslo\"}"}}`,
``,
`data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":9}}`,
``,
`data: {"type":"message_stop"}`,
``,
}, "\n")
s := newStream(io.NopCloser(strings.NewReader(sse)), "claude")
var id, name, args, finish string
for {
ev, err := s.Recv()
if err == io.EOF {
break
}
if err != nil {
t.Fatalf("Recv: %v", err)
}
var chunk struct {
Choices []struct {
Delta struct {
ToolCalls []struct {
Index int `json:"index"`
ID string `json:"id"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
} `json:"tool_calls"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
}
if err := json.Unmarshal(ev.Data, &chunk); err != nil {
t.Fatalf("chunk unmarshal: %v", err)
}
for _, tc := range chunk.Choices[0].Delta.ToolCalls {
if tc.Index != 0 {
t.Errorf("tool call index = %d, want 0", tc.Index)
}
if tc.ID != "" {
id = tc.ID
name = tc.Function.Name
}
args += tc.Function.Arguments
}
if f := chunk.Choices[0].FinishReason; f != nil {
finish = *f
}
}
if id != "toolu_1" || name != "get_weather" {
t.Errorf("tool call id=%q name=%q", id, name)
}
if args != `{"city":"Oslo"}` {
t.Errorf("arguments = %q", args)
}
if finish != "tool_calls" {
t.Errorf("finish_reason = %q", finish)
}
if u := s.Usage(); u.PromptTokens != 10 || u.CompletionTokens != 9 || u.TotalTokens != 19 {
t.Errorf("usage = %+v", u)
}
}

View File

@@ -17,10 +17,13 @@ func buildMessagesRequest(cr *providers.ChatRequest, stream bool) ([]byte, error
"stream": stream,
"max_tokens": maxTokens(p),
}
if v, ok := floatField(p, "temperature"); ok {
out["temperature"] = v
// Current Claude models reject requests that set both temperature and
// top_p, so temperature wins when a client sends both.
temp, hasTemp := floatField(p, "temperature")
if hasTemp {
out["temperature"] = temp
}
if v, ok := floatField(p, "top_p"); ok {
if v, ok := floatField(p, "top_p"); ok && !hasTemp {
out["top_p"] = v
}
if v, ok := floatField(p, "top_k"); ok {
@@ -63,7 +66,7 @@ func splitMessages(v any) (system []map[string]any, messages []map[string]any) {
"content": []map[string]any{{
"type": "tool_result",
"tool_use_id": stringField(m, "tool_call_id"),
"content": stringField(m, "content"),
"content": textContent(m["content"]),
}},
})
default: // "user" and unknown roles
@@ -88,6 +91,23 @@ func markCache(blocks []map[string]any) {
blocks[len(blocks)-1]["cache_control"] = map[string]any{"type": "ephemeral"}
}
// textContent flattens OpenAI message content (string or parts array) to text.
func textContent(v any) string {
switch c := v.(type) {
case string:
return c
case []any:
var b strings.Builder
for _, raw := range c {
if part, _ := raw.(map[string]any); part != nil && part["type"] == "text" {
b.WriteString(stringField(part, "text"))
}
}
return b.String()
}
return ""
}
// contentBlocks converts OpenAI message content (string or multimodal array)
// into Anthropic content blocks.
func contentBlocks(v any) []map[string]any {
@@ -210,9 +230,10 @@ func translateResponse(body []byte, model string) ([]byte, providers.Usage) {
} `json:"content"`
StopReason string `json:"stop_reason"`
Usage struct {
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
CacheReadInputTokens int `json:"cache_read_input_tokens"`
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
CacheReadInputTokens int `json:"cache_read_input_tokens"`
CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
} `json:"usage"`
}
_ = json.Unmarshal(body, &ar)
@@ -239,8 +260,10 @@ func translateResponse(body []byte, model string) ([]byte, providers.Usage) {
}
}
// Anthropic reports uncached, cache-read and cache-write input tokens as
// three disjoint counts. OpenAI prompt_tokens is the sum of all input.
usage := providers.Usage{
PromptTokens: ar.Usage.InputTokens + ar.Usage.CacheReadInputTokens,
PromptTokens: ar.Usage.InputTokens + ar.Usage.CacheReadInputTokens + ar.Usage.CacheCreationInputTokens,
CompletionTokens: ar.Usage.OutputTokens,
CachedTokens: ar.Usage.CacheReadInputTokens,
}

View File

@@ -2,6 +2,7 @@ package anthropic
import (
"encoding/json"
"strings"
"testing"
"git.govekar.net/LovroG05/localmodelroutex/internal/providers"
@@ -158,3 +159,51 @@ func TestMapStopReason(t *testing.T) {
}
}
}
func TestBuildMessagesRequestDropsTopPWhenTemperatureSet(t *testing.T) {
cr := &providers.ChatRequest{Model: "claude", Payload: map[string]any{
"temperature": 0.2, "top_p": 0.9,
"messages": []any{map[string]any{"role": "user", "content": "hi"}},
}}
body, err := buildMessagesRequest(cr, false)
if err != nil {
t.Fatalf("build: %v", err)
}
var out map[string]any
_ = json.Unmarshal(body, &out)
if _, ok := out["top_p"]; ok {
t.Errorf("top_p must be dropped when temperature is set: %s", body)
}
if out["temperature"] != 0.2 {
t.Errorf("temperature = %v", out["temperature"])
}
}
func TestToolResultArrayContent(t *testing.T) {
cr := &providers.ChatRequest{Model: "claude", Payload: map[string]any{
"messages": []any{
map[string]any{"role": "user", "content": "run"},
map[string]any{"role": "assistant", "tool_calls": []any{map[string]any{"id": "t1", "function": map[string]any{"name": "f", "arguments": "{}"}}}},
map[string]any{"role": "tool", "tool_call_id": "t1", "content": []any{
map[string]any{"type": "text", "text": "part one "},
map[string]any{"type": "text", "text": "part two"},
}},
},
}}
body, err := buildMessagesRequest(cr, false)
if err != nil {
t.Fatalf("build: %v", err)
}
if !strings.Contains(string(body), `"content":"part one part two"`) {
t.Errorf("tool_result content not flattened: %s", body)
}
}
func TestTranslateResponseCountsCacheCreation(t *testing.T) {
body := []byte(`{"id":"m","content":[{"type":"text","text":"ok"}],"stop_reason":"end_turn",
"usage":{"input_tokens":10,"output_tokens":5,"cache_read_input_tokens":20,"cache_creation_input_tokens":30}}`)
_, u := translateResponse(body, "claude")
if u.PromptTokens != 60 || u.CachedTokens != 20 || u.CompletionTokens != 5 || u.TotalTokens != 65 {
t.Errorf("usage = %+v", u)
}
}

View File

@@ -29,11 +29,13 @@ type provider struct {
name string
}
// New builds a Copilot provider. Credentials: {"pat": "<github token>"}.
// New builds a Copilot provider. Credentials: {"pat": "<github token>"};
// "token" and "api_key" are accepted as aliases.
func New(cfg providers.Config) (providers.Provider, error) {
var sec struct {
PAT string `json:"pat"`
Token string `json:"token"` // accepted as an alias for pat
PAT string `json:"pat"`
Token string `json:"token"`
APIKey string `json:"api_key"`
}
if err := providers.DecodeCredentials(cfg, &sec); err != nil {
return nil, err
@@ -43,7 +45,10 @@ func New(cfg providers.Config) (providers.Provider, error) {
pat = sec.Token
}
if pat == "" {
return nil, errors.New("copilot: missing github token (credentials.pat)")
pat = sec.APIKey
}
if pat == "" {
return nil, errors.New("copilot: missing github token (credentials.api_key)")
}
base := cfg.BaseURL

View File

@@ -2,6 +2,7 @@ package store
import (
"context"
"time"
"github.com/google/uuid"
)
@@ -9,28 +10,33 @@ import (
// AdminModelRow is a catalog entry for the admin view (includes blocked /
// unavailable rows, with both discovered and override pricing).
type AdminModelRow struct {
ID uuid.UUID `json:"id"`
ProviderName string `json:"provider_name"`
Slug string `json:"slug"`
DisplayName string `json:"display_name"`
Source string `json:"source"`
Available bool `json:"available"`
Enabled bool `json:"enabled"`
Blocked bool `json:"blocked"`
ContextLength *int `json:"context_length"`
DiscInputPerMTok *float64 `json:"disc_input_per_mtok"`
DiscOutputPerMTok *float64 `json:"disc_output_per_mtok"`
PriceInputPerMTok *float64 `json:"price_input_per_mtok"`
PriceOutputPerMTok *float64 `json:"price_output_per_mtok"`
ID uuid.UUID `json:"id"`
ProviderName string `json:"provider_name"`
Slug string `json:"slug"`
DisplayName string `json:"display_name"`
Source string `json:"source"`
Available bool `json:"available"`
Enabled bool `json:"enabled"`
Blocked bool `json:"blocked"`
ContextLength *int `json:"context_length"`
LastSeenAt *time.Time `json:"last_seen_at"`
DiscInputPerMTok *float64 `json:"disc_input_per_mtok"`
DiscOutputPerMTok *float64 `json:"disc_output_per_mtok"`
PriceInputPerMTok *float64 `json:"price_input_per_mtok"`
PriceOutputPerMTok *float64 `json:"price_output_per_mtok"`
PriceCachedPerMTok *float64 `json:"price_cached_per_mtok"`
RPMLimit *int `json:"rpm_limit"`
TPMLimit *int `json:"tpm_limit"`
}
// ListModelsAdmin lists every catalog entry (no servability filter) for admin.
func (s *Store) ListModelsAdmin(ctx context.Context) ([]AdminModelRow, error) {
rows, err := s.pool.Query(ctx, `
SELECT pm.id, p.name, pm.slug, COALESCE(pm.display_name,''), pm.source,
pm.available, pm.enabled, pm.blocked, pm.context_length,
pm.available, pm.enabled, pm.blocked, pm.context_length, pm.last_seen_at,
pm.disc_input_price_per_mtok, pm.disc_output_price_per_mtok,
pm.price_input_per_mtok, pm.price_output_per_mtok
pm.price_input_per_mtok, pm.price_output_per_mtok, pm.price_cached_per_mtok,
pm.rpm_limit, pm.tpm_limit
FROM provider_models pm
JOIN providers p ON p.id = pm.provider_id
ORDER BY p.name, pm.slug`)
@@ -42,9 +48,10 @@ func (s *Store) ListModelsAdmin(ctx context.Context) ([]AdminModelRow, error) {
for rows.Next() {
var m AdminModelRow
if err := rows.Scan(&m.ID, &m.ProviderName, &m.Slug, &m.DisplayName, &m.Source,
&m.Available, &m.Enabled, &m.Blocked, &m.ContextLength,
&m.Available, &m.Enabled, &m.Blocked, &m.ContextLength, &m.LastSeenAt,
&m.DiscInputPerMTok, &m.DiscOutputPerMTok,
&m.PriceInputPerMTok, &m.PriceOutputPerMTok); err != nil {
&m.PriceInputPerMTok, &m.PriceOutputPerMTok, &m.PriceCachedPerMTok,
&m.RPMLimit, &m.TPMLimit); err != nil {
return nil, err
}
out = append(out, m)
@@ -52,16 +59,23 @@ func (s *Store) ListModelsAdmin(ctx context.Context) ([]AdminModelRow, error) {
return out, rows.Err()
}
// ModelOverride carries admin override fields; nil fields are left unchanged.
// Patch is an optional column update. Set=false leaves the column unchanged.
// Set=true with a nil Value clears the column to NULL.
type Patch[T any] struct {
Set bool
Value *T
}
// ModelOverride carries admin override fields; unset patches are left unchanged.
type ModelOverride struct {
Enabled *bool
Blocked *bool
ParamOverrides []byte // jsonb; nil leaves unchanged
InputPrice *float64
OutputPrice *float64
CachedPrice *float64
RPMLimit *int
TPMLimit *int
InputPrice Patch[float64]
OutputPrice Patch[float64]
CachedPrice Patch[float64]
RPMLimit Patch[int]
TPMLimit Patch[int]
}
func (s *Store) UpdateModelOverride(ctx context.Context, id uuid.UUID, o ModelOverride) error {
@@ -74,13 +88,18 @@ func (s *Store) UpdateModelOverride(ctx context.Context, id uuid.UUID, o ModelOv
enabled = COALESCE($2, enabled),
blocked = COALESCE($3, blocked),
param_overrides = COALESCE($4, param_overrides),
price_input_per_mtok = COALESCE($5, price_input_per_mtok),
price_output_per_mtok = COALESCE($6, price_output_per_mtok),
price_cached_per_mtok = COALESCE($7, price_cached_per_mtok),
rpm_limit = COALESCE($8, rpm_limit),
tpm_limit = COALESCE($9, tpm_limit)
price_input_per_mtok = CASE WHEN $5::boolean THEN $6 ELSE price_input_per_mtok END,
price_output_per_mtok = CASE WHEN $7::boolean THEN $8 ELSE price_output_per_mtok END,
price_cached_per_mtok = CASE WHEN $9::boolean THEN $10 ELSE price_cached_per_mtok END,
rpm_limit = CASE WHEN $11::boolean THEN $12 ELSE rpm_limit END,
tpm_limit = CASE WHEN $13::boolean THEN $14 ELSE tpm_limit END
WHERE id = $1`,
id, o.Enabled, o.Blocked, params, o.InputPrice, o.OutputPrice, o.CachedPrice, o.RPMLimit, o.TPMLimit,
id, o.Enabled, o.Blocked, params,
o.InputPrice.Set, o.InputPrice.Value,
o.OutputPrice.Set, o.OutputPrice.Value,
o.CachedPrice.Set, o.CachedPrice.Value,
o.RPMLimit.Set, o.RPMLimit.Value,
o.TPMLimit.Set, o.TPMLimit.Value,
)
return err
}

View File

@@ -2,6 +2,7 @@ package store
import (
"context"
"time"
"github.com/google/uuid"
)
@@ -15,11 +16,12 @@ type ProviderRow struct {
Credentials []byte // encrypted (nonce||ciphertext); may be nil
Enabled bool
Config []byte // jsonb
CreatedAt time.Time
}
// ListProviders returns configured providers, optionally only enabled ones.
func (s *Store) ListProviders(ctx context.Context, enabledOnly bool) ([]ProviderRow, error) {
q := `SELECT id, name, type, COALESCE(base_url,''), credentials, enabled, config FROM providers`
q := `SELECT id, name, type, COALESCE(base_url,''), credentials, enabled, config, created_at FROM providers`
if enabledOnly {
q += ` WHERE enabled = true`
}
@@ -34,7 +36,7 @@ func (s *Store) ListProviders(ctx context.Context, enabledOnly bool) ([]Provider
var out []ProviderRow
for rows.Next() {
var p ProviderRow
if err := rows.Scan(&p.ID, &p.Name, &p.Type, &p.BaseURL, &p.Credentials, &p.Enabled, &p.Config); err != nil {
if err := rows.Scan(&p.ID, &p.Name, &p.Type, &p.BaseURL, &p.Credentials, &p.Enabled, &p.Config, &p.CreatedAt); err != nil {
return nil, err
}
out = append(out, p)
@@ -66,6 +68,32 @@ func (s *Store) CreateProvider(ctx context.Context, p CreateProviderParams) (uui
return id, err
}
// UpdateProviderParams carries the editable provider fields. Nil fields are
// left unchanged. Credentials must already be encrypted.
type UpdateProviderParams struct {
BaseURL *string
Enabled *bool
Credentials []byte
Config []byte
}
func (s *Store) UpdateProvider(ctx context.Context, id uuid.UUID, p UpdateProviderParams) error {
var baseURL any
if p.BaseURL != nil {
baseURL = nullString(*p.BaseURL)
}
_, err := s.pool.Exec(ctx, `
UPDATE providers SET
base_url = CASE WHEN $2::boolean THEN $3 ELSE base_url END,
enabled = COALESCE($4, enabled),
credentials = COALESCE($5, credentials),
config = COALESCE($6, config)
WHERE id = $1`,
id, p.BaseURL != nil, baseURL, p.Enabled, p.Credentials, p.Config,
)
return err
}
func (s *Store) SetProviderEnabled(ctx context.Context, id uuid.UUID, enabled bool) error {
_, err := s.pool.Exec(ctx, `UPDATE providers SET enabled = $2 WHERE id = $1`, id, enabled)
return err
@@ -75,3 +103,32 @@ func (s *Store) DeleteProvider(ctx context.Context, id uuid.UUID) error {
_, err := s.pool.Exec(ctx, `DELETE FROM providers WHERE id = $1`, id)
return err
}
// ProviderModelCount summarizes a provider's catalog rows.
type ProviderModelCount struct {
Models int
Available int
LastSyncedAt *time.Time
}
// ProviderModelCounts returns per-provider catalog counts keyed by provider id.
func (s *Store) ProviderModelCounts(ctx context.Context) (map[uuid.UUID]ProviderModelCount, error) {
rows, err := s.pool.Query(ctx, `
SELECT provider_id, COUNT(*)::int, COUNT(*) FILTER (WHERE available)::int, MAX(last_seen_at)
FROM provider_models
GROUP BY provider_id`)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[uuid.UUID]ProviderModelCount{}
for rows.Next() {
var id uuid.UUID
var c ProviderModelCount
if err := rows.Scan(&id, &c.Models, &c.Available, &c.LastSyncedAt); err != nil {
return nil, err
}
out[id] = c
}
return out, rows.Err()
}

View File

@@ -82,6 +82,7 @@ type RequestLogListRow struct {
Error string
PromptTokens int
CompletionTokens int
CachedTokens int
TotalTokens int
Cost float64
LatencyMS int
@@ -133,7 +134,7 @@ func (s *Store) ListRequestLogs(ctx context.Context, p ListLogsParams) ([]Reques
SELECT rl.id::text, rl.created_at, COALESCE(rl.request_id,''),
COALESCE(rl.key_prefix,''), COALESCE(rl.model_name,''), COALESCE(rl.provider_name,''),
rl.status_code, rl.streamed, COALESCE(rl.error,''),
rl.prompt_tokens, rl.completion_tokens, rl.total_tokens,
rl.prompt_tokens, rl.completion_tokens, rl.cached_tokens, rl.total_tokens,
rl.cost, rl.latency_ms, rl.ttft_ms
FROM request_logs rl
WHERE ($1 = '' OR rl.provider_name = $1)
@@ -162,7 +163,7 @@ func (s *Store) ListRequestLogs(ctx context.Context, p ListLogsParams) ([]Reques
&r.ID, &r.CreatedAt, &r.RequestID,
&r.KeyPrefix, &r.ModelName, &r.ProviderName,
&r.StatusCode, &r.Streamed, &r.Error,
&r.PromptTokens, &r.CompletionTokens, &r.TotalTokens,
&r.PromptTokens, &r.CompletionTokens, &r.CachedTokens, &r.TotalTokens,
&r.Cost, &r.LatencyMS, &r.TTFTMs,
); err != nil {
return nil, err
@@ -250,7 +251,7 @@ type LogsSummary struct {
}
// GetLogsSummary computes aggregate totals since the given time in a single
// query, instead of shipping raw rows to the client to sum in JS.
// query. Every aggregate is coalesced so an empty window scans as zeros.
func (s *Store) GetLogsSummary(ctx context.Context, since time.Time, userID *uuid.UUID) (LogsSummary, error) {
var out LogsSummary
err := s.pool.QueryRow(ctx, `
@@ -258,7 +259,7 @@ func (s *Store) GetLogsSummary(ctx context.Context, since time.Time, userID *uui
COALESCE(SUM(total_tokens),0)::bigint,
COALESCE(SUM(cost),0),
COALESCE(AVG(latency_ms),0)::int,
SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END)::int
COALESCE(SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END),0)::int
FROM request_logs
WHERE created_at >= $1
AND ($2::uuid IS NULL OR user_id = $2)`,

View File

@@ -0,0 +1,184 @@
package store_test
import (
"context"
"os"
"testing"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
"git.govekar.net/LovroG05/localmodelroutex/internal/store"
)
// TestStoreQueries exercises the hand-written SQL against a real Postgres with
// migrations applied. Set GATEWAY_TEST_DB_URL to run.
func TestStoreQueries(t *testing.T) {
url := os.Getenv("GATEWAY_TEST_DB_URL")
if url == "" {
t.Skip("set GATEWAY_TEST_DB_URL to run the store DB integration test")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, url)
if err != nil {
t.Fatalf("connect: %v", err)
}
defer pool.Close()
for _, table := range []string{"request_logs", "virtual_keys", "providers", "users"} {
if _, err := pool.Exec(ctx, "DELETE FROM "+table); err != nil {
t.Fatalf("cleanup %s: %v", table, err)
}
}
st := store.New(pool)
// Empty window: every aggregate must scan as zero, not NULL.
sum, err := st.GetLogsSummary(ctx, time.Now().Add(-time.Hour), nil)
if err != nil {
t.Fatalf("GetLogsSummary on empty table: %v", err)
}
if sum.Requests != 0 || sum.Errors != 0 || sum.TotalCost != 0 {
t.Fatalf("empty summary = %+v", sum)
}
user, err := st.UpsertUser(ctx, "sub-1", "a@example.com", "A")
if err != nil {
t.Fatalf("UpsertUser: %v", err)
}
// Keys: rules and created_at come back from the list query.
keyID, err := st.CreateVirtualKey(ctx, store.CreateVirtualKeyParams{
Name: "k", KeyHash: []byte("hash-1"), KeyPrefix: "sk-00000001", UserID: &user.ID,
Rules: []string{"local/*", "openai/gpt-4o"},
})
if err != nil {
t.Fatalf("CreateVirtualKey: %v", err)
}
keys, err := st.ListVirtualKeys(ctx, &user.ID)
if err != nil {
t.Fatalf("ListVirtualKeys: %v", err)
}
if len(keys) != 1 || len(keys[0].Rules) != 2 || keys[0].Rules[0] != "local/*" || keys[0].CreatedAt.IsZero() {
t.Fatalf("keys = %+v", keys)
}
// Providers: update, counts, and the catalog join.
provID, err := st.CreateProvider(ctx, store.CreateProviderParams{Name: "local", Type: "llamacpp", Enabled: true})
if err != nil {
t.Fatalf("CreateProvider: %v", err)
}
newURL := "http://10.0.0.5:8080/v1"
off := false
if err := st.UpdateProvider(ctx, provID, store.UpdateProviderParams{BaseURL: &newURL, Enabled: &off, Config: []byte(`{"a":"b"}`)}); err != nil {
t.Fatalf("UpdateProvider: %v", err)
}
// Other DB tests may run against the same database, so look the row up by id.
findProv := func() store.ProviderRow {
provs, err := st.ListProviders(ctx, false)
if err != nil {
t.Fatalf("ListProviders: %v", err)
}
for _, p := range provs {
if p.ID == provID {
return p
}
}
t.Fatalf("provider %s missing from %+v", provID, provs)
return store.ProviderRow{}
}
if p := findProv(); p.BaseURL != newURL || p.Enabled || string(p.Config) != `{"a": "b"}` {
t.Fatalf("provider after update = %+v", p)
}
empty := ""
if err := st.UpdateProvider(ctx, provID, store.UpdateProviderParams{BaseURL: &empty}); err != nil {
t.Fatalf("UpdateProvider clear url: %v", err)
}
if p := findProv(); p.BaseURL != "" {
t.Fatalf("base_url not cleared: %q", p.BaseURL)
}
ctxLen := 4096
if err := st.UpsertDiscovered(ctx, provID, time.Now(), []store.DiscoveredModel{
{Slug: "m1", ContextLength: &ctxLen}, {Slug: "m2"},
}); err != nil {
t.Fatalf("UpsertDiscovered: %v", err)
}
if err := st.UpsertDiscovered(ctx, provID, time.Now(), []store.DiscoveredModel{{Slug: "m1"}}); err != nil {
t.Fatalf("UpsertDiscovered 2: %v", err)
}
counts, err := st.ProviderModelCounts(ctx)
if err != nil {
t.Fatalf("ProviderModelCounts: %v", err)
}
if c := counts[provID]; c.Models != 2 || c.Available != 1 || c.LastSyncedAt == nil {
t.Fatalf("counts = %+v", c)
}
// Overrides: set, then clear with an explicit null patch.
rows, err := st.ListModelsAdmin(ctx)
if err != nil {
t.Fatalf("ListModelsAdmin: %v", err)
}
var m1 uuid.UUID
for _, r := range rows {
if r.ProviderName == "local" && r.Slug == "m1" {
m1 = r.ID
}
}
if m1 == uuid.Nil {
t.Fatalf("local/m1 missing from %+v", rows)
}
price := 2.5
rpm := 30
if err := st.UpdateModelOverride(ctx, m1, store.ModelOverride{
InputPrice: store.Patch[float64]{Set: true, Value: &price},
RPMLimit: store.Patch[int]{Set: true, Value: &rpm},
}); err != nil {
t.Fatalf("UpdateModelOverride set: %v", err)
}
rows, _ = st.ListModelsAdmin(ctx)
for _, r := range rows {
if r.ID == m1 && (r.PriceInputPerMTok == nil || *r.PriceInputPerMTok != 2.5 || r.RPMLimit == nil || *r.RPMLimit != 30) {
t.Fatalf("override not applied: %+v", r)
}
}
if err := st.UpdateModelOverride(ctx, m1, store.ModelOverride{InputPrice: store.Patch[float64]{Set: true}}); err != nil {
t.Fatalf("UpdateModelOverride clear: %v", err)
}
rows, _ = st.ListModelsAdmin(ctx)
for _, r := range rows {
if r.ID == m1 {
if r.PriceInputPerMTok != nil {
t.Fatalf("price not cleared: %+v", r)
}
if r.RPMLimit == nil || *r.RPMLimit != 30 {
t.Fatalf("unset patch must leave rpm_limit alone: %+v", r)
}
}
}
// Logs: insert, then list and summarize.
logID, err := st.InsertRequestLog(ctx, store.RequestLog{
VirtualKeyID: &keyID, UserID: &user.ID, ProviderID: &provID, ProviderModelID: &m1,
KeyPrefix: "sk-00000001", ModelName: "local/m1", ProviderName: "local",
StatusCode: 502, Error: "upstream down", LatencyMS: 12, ClientIP: "127.0.0.1",
})
if err != nil {
t.Fatalf("InsertRequestLog: %v", err)
}
if err := st.InsertLogContent(ctx, logID, []byte(`[{"role":"user","content":"hi"}]`), ""); err != nil {
t.Fatalf("InsertLogContent: %v", err)
}
list, err := st.ListRequestLogs(ctx, store.ListLogsParams{UserID: &user.ID, Query: "local/m1", StatusGte: 400})
if err != nil || len(list) != 1 || list[0].Error != "upstream down" {
t.Fatalf("ListRequestLogs: %v %+v", err, list)
}
sum, err = st.GetLogsSummary(ctx, time.Now().Add(-time.Hour), &user.ID)
if err != nil || sum.Requests != 1 || sum.Errors != 1 {
t.Fatalf("summary = %+v err=%v", sum, err)
}
series, err := st.RequestVolumeSeries(ctx, time.Now().Add(-time.Hour), &user.ID, 6)
if err != nil || len(series) != 6 || series[5] != 1 {
t.Fatalf("series = %v err=%v", series, err)
}
}

View File

@@ -21,6 +21,7 @@ type VirtualKey struct {
ExpiresAt *time.Time
RevokedAt *time.Time
LastUsedAt *time.Time
CreatedAt time.Time
Rules []string
}
@@ -91,14 +92,17 @@ func (s *Store) CreateVirtualKey(ctx context.Context, p CreateVirtualKeyParams)
return id, tx.Commit(ctx)
}
// ListVirtualKeys returns virtual keys. When userID is non-nil only that user's
// keys are returned; pass nil to list all (admin view).
// ListVirtualKeys returns virtual keys with their allowlist rules. When userID
// is non-nil only that user's keys are returned; pass nil to list all (admin view).
func (s *Store) ListVirtualKeys(ctx context.Context, userID *uuid.UUID) ([]VirtualKey, error) {
rows, err := s.pool.Query(ctx, `
SELECT id, COALESCE(name,''), key_prefix, type, user_id, allow_all_models, expires_at, revoked_at, last_used_at
FROM virtual_keys
WHERE ($1::uuid IS NULL OR user_id = $1)
ORDER BY created_at DESC`, userID)
SELECT vk.id, COALESCE(vk.name,''), vk.key_prefix, vk.type, vk.user_id, vk.allow_all_models,
vk.expires_at, vk.revoked_at, vk.last_used_at, vk.created_at,
COALESCE((SELECT array_agg(r.pattern ORDER BY r.pattern)
FROM virtual_key_model_rules r WHERE r.virtual_key_id = vk.id), '{}')
FROM virtual_keys vk
WHERE ($1::uuid IS NULL OR vk.user_id = $1)
ORDER BY vk.created_at DESC`, userID)
if err != nil {
return nil, err
}
@@ -107,7 +111,7 @@ func (s *Store) ListVirtualKeys(ctx context.Context, userID *uuid.UUID) ([]Virtu
for rows.Next() {
var k VirtualKey
if err := rows.Scan(&k.ID, &k.Name, &k.KeyPrefix, &k.Type, &k.UserID, &k.AllowAllModels,
&k.ExpiresAt, &k.RevokedAt, &k.LastUsedAt); err != nil {
&k.ExpiresAt, &k.RevokedAt, &k.LastUsedAt, &k.CreatedAt, &k.Rules); err != nil {
return nil, err
}
out = append(out, k)