fix(pricing): 修复价格同步与完整度边界

This commit is contained in:
tbphp committed 2026-08-03 11:09:27 +08:00
1 parent c25c3525b5
commit 6ae94cd147
38 files changed
+808 -71

No files matched your search

+7
View File
@@ -15,6 +15,13 @@ var officialProviderIDs = map[string]struct{}{
"google": {},
}
// IsOfficialProviderID reports whether providerID is one of the built-in
// provider presets that remain selectable without a loaded catalog snapshot.
func IsOfficialProviderID(providerID string) bool {
_, ok := officialProviderIDs[providerID]
return ok
}
// OfficialSuggestions returns the three local provider presets with safe
// field-level name and API URL updates from the current catalog.
func OfficialSuggestions(snapshot *Snapshot) []Provider {
+13
View File
@@ -56,6 +56,19 @@ func TestOfficialSuggestionsReturnFallbacksForNilOrEmptySnapshot(t *testing.T) {
}
}
func TestIsOfficialProviderID(t *testing.T) {
for _, providerID := range []string{"openai", "anthropic", "google"} {
if !IsOfficialProviderID(providerID) {
t.Fatalf("IsOfficialProviderID(%q) = false, want true", providerID)
}
}
for _, providerID := range []string{"", "OpenAI", "private-company"} {
if IsOfficialProviderID(providerID) {
t.Fatalf("IsOfficialProviderID(%q) = true, want false", providerID)
}
}
}
func TestSearchProvidersExcludesOfficialSortsAndInfersConservatively(t *testing.T) {
snapshot := &Snapshot{Providers: map[string]Provider{
"openai": {ID: "openai", Name: "OpenAI"},
+2 -4
View File
@@ -66,10 +66,8 @@ func (s *Service) EnsureInitialState(ctx context.Context) error {
if err := reconcileReferencedPrices(tx, catalogSnapshot); err != nil {
return err
}
if catalogSnapshot != nil {
if err := cleanupUnreferencedAutomaticPrices(tx); err != nil {
return err
}
if err := cleanupUnreferencedAutomaticPrices(tx); err != nil {
return err
}
var err error
priceTable, err = loadPriceTable(ctx, tx)
@@ -7,6 +7,7 @@ import (
"gpt-load/internal/catalog"
"gpt-load/internal/dialect"
"gpt-load/internal/pricing"
"gpt-load/internal/protocol"
"gpt-load/internal/state"
"gpt-load/internal/storage/models"
@@ -17,9 +18,12 @@ func TestDraftDiscoveryMergesLiveAndLocalCatalogByExactIDWithoutURLInference(t *
fixture.catalogRuntime.Publish(&catalog.Snapshot{Providers: map[string]catalog.Provider{
"openai": {
ID: "openai", Name: "OpenAI", Models: map[string]catalog.Model{
"shared": {ID: "shared", Name: "Shared display"},
"catalog-only": {ID: "catalog-only", Name: "Catalog only"},
"SHARED": {ID: "SHARED", Name: "Case distinct"},
"shared": {ID: "shared", Name: "Shared display"},
"catalog-only": {
ID: "catalog-only", Name: "Catalog only",
Cost: &catalog.ModelCost{Prices: pricing.Prices{Input: priceTestValue(1)}},
},
"SHARED": {ID: "SHARED", Name: "Case distinct"},
},
},
}})
@@ -50,7 +54,7 @@ func TestDraftDiscoveryMergesLiveAndLocalCatalogByExactIDWithoutURLInference(t *
{ID: "shared", Name: "Shared display", Sources: []string{"live", "catalog"}, PricingStatus: PricingStatusConfigured},
{ID: "live-only", Name: "live-only", Sources: []string{"live"}, PricingStatus: PricingStatusPending},
{ID: "SHARED", Name: "Case distinct", Sources: []string{"catalog"}, PricingStatus: PricingStatusPending},
{ID: "catalog-only", Name: "Catalog only", Sources: []string{"catalog"}, PricingStatus: PricingStatusPending},
{ID: "catalog-only", Name: "Catalog only", Sources: []string{"catalog"}, PricingStatus: PricingStatusConfigured},
}
if !reflect.DeepEqual(got.Models, want) {
t.Fatalf("merged candidates = %#v, want %#v", got.Models, want)
@@ -77,8 +81,11 @@ func TestSavedGroupDiscoveryUsesPersistedProviderAndSharedPricingStatus(t *testi
fixture.catalogRuntime.Publish(&catalog.Snapshot{Providers: map[string]catalog.Provider{
"openai": {
ID: "openai", Models: map[string]catalog.Model{
"shared": {ID: "shared", Name: "Shared display"},
"catalog-only": {ID: "catalog-only", Name: "Catalog only"},
"shared": {ID: "shared", Name: "Shared display"},
"catalog-only": {
ID: "catalog-only", Name: "Catalog only",
Cost: &catalog.ModelCost{Prices: pricing.Prices{Input: priceTestValue(1)}},
},
},
},
}})
@@ -113,7 +120,7 @@ func TestSavedGroupDiscoveryUsesPersistedProviderAndSharedPricingStatus(t *testi
want := []ModelCandidate{
{ID: "shared", Name: "Shared display", Sources: []string{"live", "catalog"}, PricingStatus: PricingStatusConfigured},
{ID: "live-only", Name: "live-only", Sources: []string{"live"}, PricingStatus: PricingStatusPending},
{ID: "catalog-only", Name: "Catalog only", Sources: []string{"catalog"}, PricingStatus: PricingStatusPending},
{ID: "catalog-only", Name: "Catalog only", Sources: []string{"catalog"}, PricingStatus: PricingStatusConfigured},
}
if !reflect.DeepEqual(got.Models, want) {
t.Fatalf("saved group candidates = %#v, want %#v", got.Models, want)
+21 -2
View File
@@ -36,6 +36,7 @@ const (
CatalogSyncStartup CatalogSyncTrigger = "startup"
CatalogSyncPeriodic CatalogSyncTrigger = "periodic"
CatalogSyncGroup CatalogSyncTrigger = "group_change"
CatalogSyncSettings CatalogSyncTrigger = "settings_change"
CatalogSyncManual CatalogSyncTrigger = "manual"
)
@@ -127,6 +128,7 @@ type CatalogSyncCoordinator struct {
newTimer func(time.Duration) catalogSyncTimer
now func() time.Time
groupWake chan struct{}
immediateWake chan struct{}
}
var errCatalogSyncCoordinatorStopped = errors.New("catalog sync coordinator is shutting down")
@@ -148,8 +150,9 @@ func newCatalogSyncCoordinator(
newTimer: func(interval time.Duration) catalogSyncTimer {
return standardCatalogSyncTimer{timer: time.NewTimer(interval)}
},
now: time.Now,
groupWake: make(chan struct{}, 1),
now: time.Now,
groupWake: make(chan struct{}, 1),
immediateWake: make(chan struct{}, 1),
}
if service != nil {
coordinator.applySnapshot = service.applyCatalogSnapshot
@@ -235,6 +238,8 @@ func (coordinator *CatalogSyncCoordinator) Run(ctx context.Context) {
stopTimer(debounce)
debounce = coordinator.newTimer(modelsDevGroupDebounce)
debounceC = debounce.C()
case <-coordinator.immediateWake:
launch(CatalogSyncSettings)
case <-debounceC:
debounce = nil
debounceC = nil
@@ -285,6 +290,17 @@ func (coordinator *CatalogSyncCoordinator) RequestGroupSync() {
}
}
func (coordinator *CatalogSyncCoordinator) RequestImmediateSync() {
if coordinator == nil || coordinator.service == nil ||
!coordinator.service.modelsDevAutoSyncEnabled() {
return
}
select {
case coordinator.immediateWake <- struct{}{}:
default:
}
}
func (coordinator *CatalogSyncCoordinator) hasLastKnownGood() bool {
coordinator.mu.Lock()
hasLKG := coordinator.hasLKG
@@ -494,6 +510,9 @@ func (s *Service) applyCatalogSnapshot(ctx context.Context, snapshot *catalog.Sn
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
if err := s.enforceOperationRecoveryBarrierLocked(ctx, 0); err != nil {
return err
}
var table *pricing.Table
err := s.withControlTransaction(ctx, func(tx *gorm.DB) error {
+41
View File
@@ -495,6 +495,47 @@ func TestCatalogSyncSchedulerDebouncesGroupChangeBursts(t *testing.T) {
}
}
func TestCatalogSyncSchedulerRunsImmediateSettingsTrigger(t *testing.T) {
fixture := newServiceFixture(t)
calls := make(chan struct{}, 2)
client := catalogSyncClientFunc(func(context.Context, catalog.Metadata) (catalog.SyncResult, error) {
calls <- struct{}{}
return catalog.SyncResult{
Metadata: catalog.Metadata{CheckedAtMillis: 10, SuccessfulFetchAtMillis: 1},
NotModified: true,
}, nil
})
coordinator := newCatalogSyncCoordinator(fixture.service, client, "unused", catalog.Metadata{
CheckedAtMillis: 1, SuccessfulFetchAtMillis: 1,
}, true)
fixture.catalogRuntime.Publish(&catalog.Snapshot{Providers: map[string]catalog.Provider{}})
periodic := newFakeRuntimeTicker()
coordinator.newTicker = func(time.Duration) runtimeTicker { return periodic }
ctx, cancel := context.WithCancel(t.Context())
done := make(chan struct{})
go func() {
coordinator.Run(ctx)
close(done)
}()
awaitCatalogCall(t, calls)
waitForCatalogIdle(t, coordinator)
coordinator.RequestImmediateSync()
awaitCatalogCall(t, calls)
waitForCatalogIdle(t, coordinator)
if coordinator.last.Trigger != CatalogSyncSettings {
t.Fatalf("immediate trigger = %q, want %q", coordinator.last.Trigger, CatalogSyncSettings)
}
cancel()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("catalog scheduler did not stop")
}
}
func TestCatalogSyncSchedulerRetries304WithoutLKGAfterOneHour(t *testing.T) {
fixture := newServiceFixture(t)
calls := make(chan struct{}, 2)
+8 -3
View File
@@ -107,11 +107,16 @@ func (s *Service) mergeDiscoveredModels(
result := make([]ModelCandidate, 0, len(live)+len(providerModels))
seen := make(map[string]int, len(live)+len(providerModels))
for _, id := range live {
var catalogModel *catalog.Model
model, catalogMatch := providerModels[id]
if catalogMatch {
catalogModel = &model
}
candidate := ModelCandidate{
ID: id, Name: id, Sources: []string{"live"},
PricingStatus: resolvePricingStatus(rows[id]),
PricingStatus: resolveCandidatePricingStatus(rows[id], catalogModel),
}
if model, exists := providerModels[id]; exists {
if catalogMatch {
if name := strings.TrimSpace(model.Name); name != "" {
candidate.Name = name
}
@@ -131,7 +136,7 @@ func (s *Service) mergeDiscoveredModels(
}
catalogOnly = append(catalogOnly, ModelCandidate{
ID: id, Name: name, Sources: []string{"catalog"},
PricingStatus: resolvePricingStatus(rows[id]),
PricingStatus: resolveCandidatePricingStatus(rows[id], &model),
})
}
sort.Slice(catalogOnly, func(left, right int) bool {
+3
View File
@@ -156,6 +156,9 @@ func (s *Service) normalizeGroupCreate(
if err != nil {
return normalizedGroupCreate{}, app_errors.ErrValidation
}
if err := s.validateSelectableProviderID(providerID); err != nil {
return normalizedGroupCreate{}, err
}
if !request.Models.Set {
return normalizedGroupCreate{}, app_errors.ErrValidation
}
+33
View File
@@ -14,6 +14,7 @@ import (
"github.com/sirupsen/logrus"
"gpt-load/internal/catalog"
"gpt-load/internal/dialect"
"gpt-load/internal/platform/encryption"
app_errors "gpt-load/internal/platform/errors"
@@ -574,6 +575,38 @@ func TestCreateGroupRequiresModelsAndAllowsEmptyArrayAtomically(t *testing.T) {
}
}
func TestCreateGroupRequiresSelectableProviderID(t *testing.T) {
fixture := newServiceFixture(t)
providerID := "private-company"
request := GroupCreateRequest{
ProviderID: &providerID,
UpstreamURL: "https://private-company.example/v1",
Protocols: []protocol.Protocol{protocol.OpenAICompletions},
Models: optionalGroupModels{Set: true, Values: []GroupModel{}},
Keys: "sk-private-company",
}
if _, err := fixture.service.CreateGroup(t.Context(), request); !errors.Is(err, app_errors.ErrValidation) {
t.Fatalf("CreateGroup(unknown provider) error = %v, want ErrValidation", err)
}
assertGroupCount(t, fixture.db, 0)
fixture.catalogRuntime.Publish(&catalog.Snapshot{Providers: map[string]catalog.Provider{
providerID: {ID: providerID},
}})
result, err := fixture.service.CreateGroup(t.Context(), request)
if err != nil {
t.Fatalf("CreateGroup(catalog provider) error = %v", err)
}
var stored models.Group
if err := fixture.db.First(&stored, result.GroupID).Error; err != nil {
t.Fatal(err)
}
if stored.ProviderID == nil || *stored.ProviderID != providerID {
t.Fatalf("stored provider_id = %v, want %q", stored.ProviderID, providerID)
}
}
func TestCreateGroupRejectsInvalidNormalizedInputs(t *testing.T) {
fixture := newServiceFixture(t)
blank := " "
+3
View File
@@ -250,6 +250,9 @@ func (s *Service) UpdateGroupSettings(
}
if normalized.providerIDSet {
if !optionalStringsEqual(group.ProviderID, normalized.providerID) {
if err := s.validateSelectableProviderID(normalized.providerID); err != nil {
return err
}
var groupModels []GroupModel
if err := decodeGroupDiscoveryJSON(group.Models, &groupModels); err != nil {
return fmt.Errorf("decode group %d models: %w", groupID, app_errors.ErrInternalServer)
@@ -6,10 +6,12 @@ import (
"encoding/json"
"errors"
"fmt"
"reflect"
"strconv"
"testing"
"time"
"gpt-load/internal/catalog"
app_errors "gpt-load/internal/platform/errors"
"gpt-load/internal/protocol"
"gpt-load/internal/state"
@@ -177,6 +179,28 @@ func TestOperationBarrierBlocksNewMutationBehindOlderRecovery(t *testing.T) {
if ordinaryMutations != 0 {
t.Fatalf("ordinary control mutation ran behind failed barrier %d times", ordinaryMutations)
}
beforeCatalog := fixture.catalogRuntime.Load()
err = fixture.service.applyCatalogSnapshot(t.Context(), &catalog.Snapshot{
Providers: map[string]catalog.Provider{},
})
assertAPIErrorCode(t, err, app_errors.ErrControlRecoveryPending.Code)
if afterCatalog := fixture.catalogRuntime.Load(); !reflect.DeepEqual(afterCatalog, beforeCatalog) {
t.Fatalf("catalog apply published behind recovery barrier: before=%#v after=%#v", beforeCatalog, afterCatalog)
}
fixture.service.reconcileRegistryGroup = func(uint, []state.KeyEntry) (bool, error) {
return true, nil
}
recoveredCatalog := &catalog.Snapshot{Providers: map[string]catalog.Provider{
"openai": {ID: "openai"},
}}
if err := fixture.service.applyCatalogSnapshot(t.Context(), recoveredCatalog); err != nil {
t.Fatalf("catalog apply after recovery error = %v", err)
}
if afterCatalog := fixture.catalogRuntime.Load(); !reflect.DeepEqual(afterCatalog, recoveredCatalog) {
t.Fatalf("catalog apply after recovery = %#v, want %#v", afterCatalog, recoveredCatalog)
}
}
func TestOperationBarrierReturnsDatabaseErrorWhenRecoveryQueryFails(t *testing.T) {
+134 -6
View File
@@ -517,7 +517,54 @@ func TestProviderIDCanBeExplicitlyClearedWithoutURLInference(t *testing.T) {
}
}
func TestDeleteGroupRemovesCustomScopeAndRetainsProviderScope(t *testing.T) {
func TestProviderChangeRequiresSelectableProviderButAllowsPersistedMissingProvider(t *testing.T) {
fixture := newServiceFixture(t)
legacyProviderID := "legacy-provider"
group := createPriceTestGroup(t, fixture.db, models.Group{
Name: "legacy",
ProviderID: &legacyProviderID,
UpstreamURL: "https://legacy.example/v1",
Protocols: models.JSON(`["openai-completions"]`),
Models: models.JSON(`[]`),
Config: models.JSON(`{}`),
Enabled: true,
})
if _, err := fixture.manager.Publish(mustBuildCompileInput(t, fixture.db)); err != nil {
t.Fatal(err)
}
got, err := fixture.service.UpdateGroupSettings(t.Context(), group.ID, GroupSettingsUpdateRequest{
Name: optionalField[string]{Set: true, Value: "legacy renamed"},
})
if err != nil {
t.Fatalf("UpdateGroupSettings(existing missing provider) error = %v", err)
}
if got.ProviderID == nil || *got.ProviderID != legacyProviderID {
t.Fatalf("persisted provider_id = %v, want %q", got.ProviderID, legacyProviderID)
}
unknownProviderID := "unknown-provider"
if _, err := fixture.service.UpdateGroupSettings(t.Context(), group.ID, GroupSettingsUpdateRequest{
ProviderID: optionalField[string]{Set: true, Value: unknownProviderID},
}); !errors.Is(err, app_errors.ErrValidation) {
t.Fatalf("UpdateGroupSettings(unknown provider) error = %v, want ErrValidation", err)
}
fixture.catalogRuntime.Publish(&catalog.Snapshot{Providers: map[string]catalog.Provider{
unknownProviderID: {ID: unknownProviderID},
}})
got, err = fixture.service.UpdateGroupSettings(t.Context(), group.ID, GroupSettingsUpdateRequest{
ProviderID: optionalField[string]{Set: true, Value: unknownProviderID},
})
if err != nil {
t.Fatalf("UpdateGroupSettings(catalog provider) error = %v", err)
}
if got.ProviderID == nil || *got.ProviderID != unknownProviderID {
t.Fatalf("updated provider_id = %v, want %q", got.ProviderID, unknownProviderID)
}
}
func TestDeleteGroupRemovesCustomScopeAndUnreferencedAutomaticProviderPrice(t *testing.T) {
fixture := newServiceFixture(t)
custom, err := fixture.service.CreateGroup(t.Context(), GroupCreateRequest{
UpstreamURL: "https://custom-delete.example/v1",
@@ -569,12 +616,64 @@ func TestDeleteGroupRemovesCustomScopeAndRetainsProviderScope(t *testing.T) {
Count(&providerRows).Error; err != nil {
t.Fatal(err)
}
if providerRows != 1 {
t.Fatalf("provider rows after delete = %d, want 1", providerRows)
if providerRows != 0 {
t.Fatalf("provider rows after delete = %d, want 0", providerRows)
}
}
func TestDeleteProviderGroupRemovesHistoricalCustomScopeAndRetainsProviderRows(t *testing.T) {
func TestGroupMutationCleansUnreferencedAutomaticPricesWhileAutoSyncDisabled(t *testing.T) {
fixture := newServiceFixture(t)
disabled := false
fixture.service.modelsDevAutoSyncOverride = &disabled
providerID := "openai"
created, err := fixture.service.CreateGroup(t.Context(), GroupCreateRequest{
ProviderID: &providerID,
UpstreamURL: "https://cleanup.example/v1",
Protocols: []protocol.Protocol{protocol.OpenAICompletions},
Models: optionalGroupModels{Set: true, Values: []GroupModel{
{ID: "automatic-model"},
}},
Keys: "sk-cleanup",
})
if err != nil {
t.Fatal(err)
}
manualValue := int64(5)
if err := fixture.db.Create(&models.ModelPrice{
PriceScopeKey: "provider:openai",
ModelID: "manual-unreferenced",
InputPriceNanoUSDPerMillionTokens: &manualValue,
IsManual: true,
}).Error; err != nil {
t.Fatal(err)
}
if _, err := fixture.service.UpdateGroupModels(t.Context(), created.GroupID, GroupModelsUpdateRequest{
Models: optionalGroupModels{Set: true, Values: []GroupModel{}},
}); err != nil {
t.Fatal(err)
}
var automaticCount int64
if err := fixture.db.Model(&models.ModelPrice{}).
Where("price_scope_key = ? AND model_id = ?", "provider:openai", "automatic-model").
Count(&automaticCount).Error; err != nil {
t.Fatal(err)
}
if automaticCount != 0 {
t.Fatalf("unreferenced automatic rows = %d, want 0", automaticCount)
}
var manualCount int64
if err := fixture.db.Model(&models.ModelPrice{}).
Where("price_scope_key = ? AND model_id = ?", "provider:openai", "manual-unreferenced").
Count(&manualCount).Error; err != nil {
t.Fatal(err)
}
if manualCount != 1 {
t.Fatalf("unreferenced manual rows = %d, want 1", manualCount)
}
}
func TestDeleteProviderGroupRemovesHistoricalCustomScopeAndAutomaticProviderRows(t *testing.T) {
fixture := newServiceFixture(t)
created, err := fixture.service.CreateGroup(t.Context(), GroupCreateRequest{
UpstreamURL: "https://custom-then-provider.example/v1",
@@ -629,8 +728,8 @@ func TestDeleteProviderGroupRemovesHistoricalCustomScopeAndRetainsProviderRows(t
Count(&providerRows).Error; err != nil {
t.Fatal(err)
}
if providerRows != 3 {
t.Fatalf("provider rows after delete = %d, want 3 retained", providerRows)
if providerRows != 0 {
t.Fatalf("provider rows after delete = %d, want 0", providerRows)
}
}
@@ -680,6 +779,35 @@ func TestEnsureInitialStateWithEmptyCatalogMaterializesPendingPrices(t *testing.
}
}
func TestEnsureInitialStateWithoutCatalogCleansUnreferencedAutomaticPrices(t *testing.T) {
fixture := newServiceFixture(t)
fixture.catalogRuntime = &catalog.Runtime{}
manualValue := int64(7)
for _, row := range []models.ModelPrice{
{PriceScopeKey: "provider:openai", ModelID: "automatic-orphan"},
{
PriceScopeKey: "provider:openai", ModelID: "manual-orphan",
InputPriceNanoUSDPerMillionTokens: &manualValue,
IsManual: true,
},
} {
if err := fixture.db.Create(&row).Error; err != nil {
t.Fatal(err)
}
}
if err := fixture.service.EnsureInitialState(t.Context()); err != nil {
t.Fatal(err)
}
var rows []models.ModelPrice
if err := fixture.db.Order("model_id").Find(&rows).Error; err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0].ModelID != "manual-orphan" || !rows[0].IsManual {
t.Fatalf("startup price rows = %#v, want only manual orphan", rows)
}
}
func createPriceTestGroup(t *testing.T, db *gorm.DB, group models.Group) models.Group {
t.Helper()
if err := db.Create(&group).Error; err != nil {
+22 -1
View File
@@ -1,6 +1,9 @@
package control
import "gpt-load/internal/storage/models"
import (
"gpt-load/internal/catalog"
"gpt-load/internal/storage/models"
)
type PricingStatus string
@@ -22,3 +25,21 @@ func resolvePricingStatus(row *models.ModelPrice) PricingStatus {
}
return PricingStatusPending
}
func resolveCandidatePricingStatus(
row *models.ModelPrice,
model *catalog.Model,
) PricingStatus {
if row != nil {
return resolvePricingStatus(row)
}
if model == nil || model.Cost == nil {
return PricingStatusPending
}
prices := model.Cost.Prices
if prices.Input.Set || prices.Output.Set || prices.CacheRead.Set || prices.CacheWrite.Set ||
len(model.Cost.ContextTiers) > 0 {
return PricingStatusConfigured
}
return PricingStatusPending
}
+13 -1
View File
@@ -85,6 +85,18 @@ func mapProviderSuggestion(provider catalog.Provider, official bool) ProviderSug
}
}
func (s *Service) validateSelectableProviderID(providerID *string) error {
if providerID == nil || catalog.IsOfficialProviderID(*providerID) {
return nil
}
if s.catalogRuntime != nil {
if _, exists := s.catalogRuntime.LoadProvider(*providerID); exists {
return nil
}
}
return app_errors.ErrValidation
}
func (s *Service) ListProviderModels(
ctx context.Context,
providerID string,
@@ -120,7 +132,7 @@ func (s *Service) ListProviderModels(
continue
}
row := rows[model.ID]
status := resolvePricingStatus(row)
status := resolveCandidatePricingStatus(row, &model)
if query.Status != "" && query.Status != status {
continue
}
+13 -1
View File
@@ -16,6 +16,7 @@ import (
"gpt-load/internal/catalog"
"gpt-load/internal/platform/config"
"gpt-load/internal/pricing"
"gpt-load/internal/protocol"
"gpt-load/internal/storage/models"
)
@@ -73,7 +74,7 @@ func TestProviderSuggestionsUseOfficialFallbacksMergeCatalogAndStayBounded(t *te
}
}
func TestProviderModelsAreStrictBoundedAndUsePersistedPricingStatus(t *testing.T) {
func TestProviderModelsAreStrictBoundedAndUsePersistedOrCatalogCandidatePricingStatus(t *testing.T) {
fixture := newServiceFixture(t)
providerModels := make(map[string]catalog.Model)
for index := 0; index < providerModelResultLimit+5; index++ {
@@ -81,6 +82,10 @@ func TestProviderModelsAreStrictBoundedAndUsePersistedPricingStatus(t *testing.T
providerModels[id] = catalog.Model{ID: id, Name: "Model " + id}
}
providerModels["configured"] = catalog.Model{ID: "configured", Name: "Configured"}
providerModels["catalog-priced"] = catalog.Model{
ID: "catalog-priced", Name: "Catalog Priced",
Cost: &catalog.ModelCost{Prices: pricing.Prices{Input: priceTestValue(1)}},
}
providerModels["pending"] = catalog.Model{ID: "pending", Name: "Pending"}
fixture.catalogRuntime.Publish(&catalog.Snapshot{Providers: map[string]catalog.Provider{
"openai": {ID: "openai", Name: "OpenAI", Models: providerModels},
@@ -108,6 +113,13 @@ func TestProviderModelsAreStrictBoundedAndUsePersistedPricingStatus(t *testing.T
if err != nil || len(configured.Items) != 1 || configured.Items[0].PricingStatus != PricingStatusConfigured {
t.Fatalf("configured filter = %#v, %v", configured, err)
}
catalogConfigured, err := fixture.service.ListProviderModels(t.Context(), "openai", ProviderModelQuery{
Query: "catalog-priced", Status: PricingStatusConfigured,
})
if err != nil || len(catalogConfigured.Items) != 1 ||
catalogConfigured.Items[0].PricingStatus != PricingStatusConfigured {
t.Fatalf("catalog configured filter = %#v, %v", catalogConfigured, err)
}
pending, err := fixture.service.ListProviderModels(t.Context(), "openai", ProviderModelQuery{
Query: "pending", Status: PricingStatusPending,
})
+3
View File
@@ -133,6 +133,9 @@ func (s *Service) writeGroupConfig(
if err := reconcileReferencedPrices(tx, catalogSnapshot); err != nil {
return err
}
if err := cleanupUnreferencedAutomaticPrices(tx); err != nil {
return err
}
input, err := stateloader.BuildCompileInput(ctx, tx)
if err != nil {
return err
+25 -3
View File
@@ -126,12 +126,19 @@ func (s *Service) UpdateSettings(
return SettingsResponse{}, app_errors.ErrValidation
}
_, err = s.writeConfig(ctx, func(tx *gorm.DB) error {
previousAutoSyncEnabled := false
snapshot, err := s.writeConfig(ctx, func(tx *gorm.DB) error {
previousAutoSyncEnabled = s.modelsDevAutoSyncEnabled()
return s.applySettingUpdates(tx, updates)
}, nil)
if err != nil {
return SettingsResponse{}, err
}
s.requestCatalogSyncOnEnable(
settingRequestContains(request, state.SettingModelsDevAutoSyncEnabled),
previousAutoSyncEnabled,
snapshot.Settings.ModelsDevAutoSyncEnabled,
)
return s.GetSettings(ctx)
}
@@ -156,8 +163,10 @@ func (s *Service) UpdateSettingsIfMatch(
}
var (
input state.CompileInput
result settingsWireRepresentation
input state.CompileInput
result settingsWireRepresentation
previousAutoSyncEnabled bool
updatedAutoSyncEnabled bool
)
err = s.withControlTransaction(ctx, func(tx *gorm.DB) error {
currentInput, err := stateloader.BuildCompileInput(ctx, tx)
@@ -188,6 +197,7 @@ func (s *Service) UpdateSettingsIfMatch(
},
)
}
previousAutoSyncEnabled = currentSnapshot.Settings.ModelsDevAutoSyncEnabled
if err := s.applySettingUpdates(tx, updates); err != nil {
return err
@@ -200,6 +210,7 @@ func (s *Service) UpdateSettingsIfMatch(
if err != nil {
return err
}
updatedAutoSyncEnabled = compiled.Settings.ModelsDevAutoSyncEnabled
updatedSettings, err := s.getSettingsWithSnapshot(ctx, tx, compiled)
if err != nil {
return err
@@ -215,9 +226,20 @@ func (s *Service) UpdateSettingsIfMatch(
stagePublishCommittedSnapshot,
)
}
s.requestCatalogSyncOnEnable(
settingRequestContains(request, state.SettingModelsDevAutoSyncEnabled),
previousAutoSyncEnabled,
updatedAutoSyncEnabled,
)
return result, nil
}
func (s *Service) requestCatalogSyncOnEnable(requested, before, after bool) {
if requested && !before && after && s.catalogSync != nil {
s.catalogSync.RequestImmediateSync()
}
}
func (s *Service) applySettingUpdates(
tx *gorm.DB,
updates []persistedSettingUpdate,
+86
View File
@@ -15,6 +15,7 @@ import (
"gorm.io/gorm"
"gorm.io/gorm/logger"
"gpt-load/internal/catalog"
"gpt-load/internal/platform/config"
app_errors "gpt-load/internal/platform/errors"
"gpt-load/internal/state"
@@ -100,6 +101,91 @@ func TestUpdateSettingsInjectUsageOptionsBooleanAndNullReset(t *testing.T) {
}
}
func TestUpdateSettingsEnablingModelsDevRequestsImmediateSyncOnce(t *testing.T) {
fixture := newServiceFixture(t)
if _, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{
Settings: map[string]json.RawMessage{
state.SettingModelsDevAutoSyncEnabled: json.RawMessage("false"),
},
}); err != nil {
t.Fatal(err)
}
coordinator := newCatalogSyncCoordinator(
fixture.service,
nil,
"unused",
catalog.Metadata{},
false,
)
if _, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{
Settings: map[string]json.RawMessage{
state.SettingModelsDevAutoSyncEnabled: json.RawMessage("true"),
},
}); err != nil {
t.Fatal(err)
}
select {
case <-coordinator.immediateWake:
default:
t.Fatal("false -> true did not request immediate catalog sync")
}
if _, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{
Settings: map[string]json.RawMessage{
state.SettingModelsDevAutoSyncEnabled: json.RawMessage("true"),
},
}); err != nil {
t.Fatal(err)
}
select {
case <-coordinator.immediateWake:
t.Fatal("true -> true requested another immediate catalog sync")
default:
}
}
func TestUpdateSettingsIfMatchEnablingModelsDevRequestsImmediateSync(t *testing.T) {
fixture := newServiceFixture(t)
if _, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{
Settings: map[string]json.RawMessage{
state.SettingModelsDevAutoSyncEnabled: json.RawMessage("false"),
},
}); err != nil {
t.Fatal(err)
}
current, err := fixture.service.GetSettings(t.Context())
if err != nil {
t.Fatal(err)
}
representation, err := newSettingsWireRepresentation("settings", current.DTO())
if err != nil {
t.Fatal(err)
}
coordinator := newCatalogSyncCoordinator(
fixture.service,
nil,
"unused",
catalog.Metadata{},
false,
)
if _, err := fixture.service.UpdateSettingsIfMatch(
t.Context(),
SettingsUpdateRequest{Settings: map[string]json.RawMessage{
state.SettingModelsDevAutoSyncEnabled: json.RawMessage("true"),
}},
representation.ETag,
"settings",
); err != nil {
t.Fatal(err)
}
select {
case <-coordinator.immediateWake:
default:
t.Fatal("If-Match false -> true did not request immediate catalog sync")
}
}
func TestUpdateSettingsRejectsNonBooleanInjectUsageWithoutMutation(t *testing.T) {
for _, raw := range []json.RawMessage{json.RawMessage("0"), json.RawMessage("1"), json.RawMessage(`"true"`), json.RawMessage("[]"), json.RawMessage("{}")} {
fixture := newServiceFixture(t)
+5 -3
View File
@@ -7,6 +7,7 @@ import (
"gpt-load/internal/health"
"gpt-load/internal/protocol"
"gpt-load/internal/state"
"gpt-load/internal/usage"
)
type ParsedRequest struct {
@@ -18,9 +19,10 @@ type ParsedRequest struct {
}
type RequestMetadata struct {
Model *string
Stream bool
ObserveUsage bool
Model *string
Stream bool
ObserveUsage bool
UsageDiagnostics usage.Diagnostics
}
type Dialect interface {
+1
View File
@@ -150,6 +150,7 @@ func (d *OpenAI) InspectRequest(req *ParsedRequest) (RequestMetadata, error) {
return RequestMetadata{}, fmt.Errorf("decode %s request: %w", d.Protocol(), err)
}
metadata.ObserveUsage = true
metadata.UsageDiagnostics = openAIRequestPricingDiagnostics(req.Body)
return metadata, nil
}
+61
View File
@@ -0,0 +1,61 @@
package dialect
import (
"encoding/json"
"strings"
"gpt-load/internal/usage"
)
func openAIRequestPricingDiagnostics(body []byte) usage.Diagnostics {
diagnostics := usage.Diagnostics{}
root, err := decodeJSONObject(body)
if err != nil {
return diagnostics
}
if openAIUnsupportedServiceTier(root) ||
jsonStringEquals(root, "speed", "fast") ||
openAIUnsupportedReasoningMode(root) {
diagnostics.Add(usage.DiagnosticUnsupportedBillableDetail)
}
return diagnostics
}
func openAIUnsupportedServiceTier(root map[string]json.RawMessage) bool {
value, ok := jsonString(root, "service_tier")
if !ok {
return false
}
switch strings.ToLower(value) {
case "", "auto", "default":
return false
default:
return true
}
}
func openAIUnsupportedReasoningMode(root map[string]json.RawMessage) bool {
raw, exists := root["reasoning"]
if !exists {
return false
}
object, err := decodeJSONObject(raw)
return err == nil && jsonStringEquals(object, "mode", "pro")
}
func jsonStringEquals(object map[string]json.RawMessage, field, want string) bool {
value, ok := jsonString(object, field)
return ok && strings.EqualFold(value, want)
}
func jsonString(object map[string]json.RawMessage, field string) (string, bool) {
raw, exists := object[field]
if !exists {
return "", false
}
var value string
if err := json.Unmarshal(raw, &value); err != nil {
return "", false
}
return value, true
}
+3
View File
@@ -62,6 +62,9 @@ func (d *OpenAIResponses) InspectRequest(req *ParsedRequest) (RequestMetadata, e
}
metadata.ObserveUsage = req.Method == http.MethodPost &&
(req.Path == openAIResponsesPath || req.Path == openAIResponsesCompactPath)
if len(req.Body) > 0 {
metadata.UsageDiagnostics = openAIRequestPricingDiagnostics(req.Body)
}
return metadata, nil
}
+23
View File
@@ -12,6 +12,7 @@ import (
"gpt-load/internal/health"
"gpt-load/internal/protocol"
"gpt-load/internal/state"
"gpt-load/internal/usage"
)
func TestOpenAIResponsesProtocolAndRequestMetadata(t *testing.T) {
@@ -234,6 +235,28 @@ func TestOpenAIResponsesProtocolAndRequestMetadata(t *testing.T) {
}
}
func TestOpenAIResponsesRequestMarksUnsupportedPricingModes(t *testing.T) {
t.Parallel()
selected := NewOpenAIResponses(http.DefaultClient)
for _, body := range []string{
`{"model":"gpt-5","service_tier":"priority"}`,
`{"model":"gpt-5","speed":"fast"}`,
`{"model":"gpt-5","reasoning":{"mode":"pro"}}`,
} {
metadata, err := selected.InspectRequest(&ParsedRequest{
Method: http.MethodPost,
Path: "/v1/responses",
Body: []byte(body),
})
if err != nil {
t.Fatalf("InspectRequest(%s) error = %v", body, err)
}
if !metadata.UsageDiagnostics.Has(usage.DiagnosticUnsupportedBillableDetail) {
t.Fatalf("InspectRequest(%s) diagnostics = %#v, want unsupported billable detail", body, metadata.UsageDiagnostics)
}
}
}
func TestOpenAIResponsesResource404DoesNotBlameKey(t *testing.T) {
t.Parallel()
@@ -135,6 +135,9 @@ func openAIResponsesUsagePatch(
if usageObject == nil {
return usage.Patch{Final: final, Diagnostics: diagnostics}, false
}
if openAIUnsupportedServiceTier(root) {
diagnostics.Add(usage.DiagnosticUnsupportedBillableDetail)
}
input, inputDiagnostics := usageInteger(
usageObject,
@@ -167,6 +170,20 @@ func openAIResponsesUsagePatch(
false,
)
diagnostics.Merge(detailDiagnostics)
diagnostics.Merge(usageUnsupportedPositiveIntegerDetails(details, "audio_tokens"))
} else {
diagnostics.Merge(detailDiagnostics)
}
if details, detailDiagnostics := usageOptionalObject(
usageObject,
"output_tokens_details",
); details != nil {
diagnostics.Merge(detailDiagnostics)
diagnostics.Merge(usageUnsupportedPositiveIntegerDetails(
details,
"reasoning_tokens",
"audio_tokens",
))
} else {
diagnostics.Merge(detailDiagnostics)
}
@@ -47,6 +47,24 @@ func TestOpenAIResponsesUsageNonStreaming(t *testing.T) {
want: usage.Tokens{UncachedInput: 100, Output: 30},
state: usage.StateComplete,
},
{
name: "reasoning and audio are unsupported billable details",
body: `{"usage":{"input_tokens":100,"input_tokens_details":{"cached_tokens":20,"audio_tokens":10},"output_tokens":30,"output_tokens_details":{"reasoning_tokens":12,"audio_tokens":5},"total_tokens":130}}`,
want: usage.Tokens{UncachedInput: 80, CacheRead: 20, Output: 30},
state: usage.StateComplete,
diagnostics: []usage.DiagnosticCode{
usage.DiagnosticUnsupportedBillableDetail,
},
},
{
name: "effective priority service tier is unsupported",
body: `{"service_tier":"priority","usage":{"input_tokens":100,"output_tokens":30,"total_tokens":130}}`,
want: usage.Tokens{UncachedInput: 100, Output: 30},
state: usage.StateComplete,
diagnostics: []usage.DiagnosticCode{
usage.DiagnosticUnsupportedBillableDetail,
},
},
{
name: "cached exceeds input",
body: `{"usage":{"input_tokens":100,"input_tokens_details":{"cached_tokens":120},"output_tokens":30}}`,
+33
View File
@@ -16,6 +16,7 @@ import (
"gpt-load/internal/protocol"
"gpt-load/internal/state"
"gpt-load/internal/testutil/fakeupstream"
"gpt-load/internal/usage"
)
func TestParsedRequestCarriesForwardingInputs(t *testing.T) {
@@ -222,6 +223,38 @@ func TestOpenAIInspectRequest(t *testing.T) {
}
}
func TestOpenAIInspectRequestMarksUnsupportedPricingModes(t *testing.T) {
t.Parallel()
dialect := NewOpenAI(http.DefaultClient)
for _, body := range []string{
`{"model":"gpt-5","service_tier":"priority"}`,
`{"model":"gpt-5","service_tier":"flex"}`,
`{"model":"gpt-5","speed":"fast"}`,
`{"model":"gpt-5","reasoning":{"mode":"pro"}}`,
} {
metadata, err := dialect.InspectRequest(&ParsedRequest{Body: []byte(body)})
if err != nil {
t.Fatalf("InspectRequest(%s) error = %v", body, err)
}
if !metadata.UsageDiagnostics.Has(usage.DiagnosticUnsupportedBillableDetail) {
t.Fatalf("InspectRequest(%s) diagnostics = %#v, want unsupported billable detail", body, metadata.UsageDiagnostics)
}
}
for _, body := range []string{
`{"model":"gpt-5"}`,
`{"model":"gpt-5","service_tier":"auto"}`,
`{"model":"gpt-5","service_tier":"default"}`,
} {
metadata, err := dialect.InspectRequest(&ParsedRequest{Body: []byte(body)})
if err != nil {
t.Fatalf("InspectRequest(%s) error = %v", body, err)
}
if metadata.UsageDiagnostics.Has(usage.DiagnosticUnsupportedBillableDetail) {
t.Fatalf("InspectRequest(%s) unexpectedly marked unsupported", body)
}
}
}
func TestOpenAIClassifyStatus(t *testing.T) {
dialect := NewOpenAI(http.DefaultClient)
tests := []struct {
+14
View File
@@ -80,6 +80,9 @@ func openAIUsagePatch(root map[string]json.RawMessage, final bool) (usage.Patch,
if usageObject == nil {
return usage.Patch{Diagnostics: diagnostics}, false
}
if openAIUnsupportedServiceTier(root) {
diagnostics.Add(usage.DiagnosticUnsupportedBillableDetail)
}
prompt, promptDiagnostics := usageInteger(usageObject, "prompt_tokens", true)
diagnostics.Merge(promptDiagnostics)
@@ -100,6 +103,17 @@ func openAIUsagePatch(root map[string]json.RawMessage, final bool) (usage.Patch,
cacheWriteUnknown = cacheWrite
diagnostics.Add(usage.DiagnosticUnsupportedBillableDetail)
}
diagnostics.Merge(usageUnsupportedPositiveIntegerDetails(details, "audio_tokens"))
} else {
diagnostics.Merge(detailDiagnostics)
}
if details, detailDiagnostics := usageOptionalObject(usageObject, "completion_tokens_details"); details != nil {
diagnostics.Merge(detailDiagnostics)
diagnostics.Merge(usageUnsupportedPositiveIntegerDetails(
details,
"reasoning_tokens",
"audio_tokens",
))
} else {
diagnostics.Merge(detailDiagnostics)
}
+14
View File
@@ -18,6 +18,7 @@ func TestUsageOpenAICanonicalFixtures(t *testing.T) {
if result.State != usage.StateComplete || result.Tokens != want {
t.Fatalf("non-stream result = %#v", result)
}
requireUsageDiagnostics(t, result.Diagnostics, usage.DiagnosticUnsupportedBillableDetail)
stream := extractor.NewUsageStreamExtractor()
observeUsageJSONL(t, stream, readUsageFixture(t, "openai", "stream.jsonl"))
@@ -25,6 +26,7 @@ func TestUsageOpenAICanonicalFixtures(t *testing.T) {
if !ok || streamResult.State != usage.StateComplete || streamResult.Tokens != want {
t.Fatalf("stream result = %#v, %t", streamResult, ok)
}
requireUsageDiagnostics(t, streamResult.Diagnostics, usage.DiagnosticUnsupportedBillableDetail)
if _, ok := stream.Finalize(); ok {
t.Fatal("second Finalize() succeeded")
}
@@ -56,6 +58,18 @@ func TestUsageOpenAINonStreamOptionalFields(t *testing.T) {
want: usage.Tokens{UncachedInput: 75, CacheRead: 20, CacheWriteUnknown: 5, Output: 30},
diagnostics: []usage.DiagnosticCode{usage.DiagnosticUnsupportedBillableDetail},
},
{
name: "positive input and output audio are unsupported billable details",
body: `{"usage":{"prompt_tokens":100,"completion_tokens":30,"total_tokens":130,"prompt_tokens_details":{"cached_tokens":20,"audio_tokens":10},"completion_tokens_details":{"audio_tokens":5}}}`,
want: usage.Tokens{UncachedInput: 80, CacheRead: 20, Output: 30},
diagnostics: []usage.DiagnosticCode{usage.DiagnosticUnsupportedBillableDetail},
},
{
name: "effective priority service tier is unsupported",
body: `{"service_tier":"priority","usage":{"prompt_tokens":100,"completion_tokens":30,"total_tokens":130}}`,
want: usage.Tokens{UncachedInput: 100, Output: 30},
diagnostics: []usage.DiagnosticCode{usage.DiagnosticUnsupportedBillableDetail},
},
{
name: "cache read and write exceeding prompt are diagnosed",
body: `{"usage":{"prompt_tokens":100,"completion_tokens":30,"total_tokens":130,"prompt_tokens_details":{"cached_tokens":80,"cache_write_tokens":30}}}`,
+15
View File
@@ -45,6 +45,21 @@ func usageFieldPresent(object map[string]json.RawMessage, field string) bool {
return exists && !bytes.Equal(bytes.TrimSpace(raw), []byte("null"))
}
func usageUnsupportedPositiveIntegerDetails(
object map[string]json.RawMessage,
fields ...string,
) usage.Diagnostics {
diagnostics := usage.Diagnostics{}
for _, field := range fields {
value, fieldDiagnostics := usageInteger(object, field, false)
diagnostics.Merge(fieldDiagnostics)
if value != nil && *value > 0 {
diagnostics.Add(usage.DiagnosticUnsupportedBillableDetail)
}
}
return diagnostics
}
func usageNormalizedTotal(reported *int64, tokens usage.Tokens, diagnostics *usage.Diagnostics) {
if reported == nil {
return
+1
View File
@@ -334,6 +334,7 @@ func (handler *Handler) Handle(ginContext *gin.Context) {
}
recorder.setClientModel(model)
recorder.setUsageApplicable(metadata.ObserveUsage)
recorder.setUsageDiagnostics(metadata.UsageDiagnostics)
allowedKeyIDs := make(map[uint]struct{}, len(allowedKeyRefs))
for keyID := range allowedKeyRefs {
+25 -16
View File
@@ -44,22 +44,23 @@ type frozenAttemptPricing struct {
}
type requestRecorder struct {
sink telemetry.RequestLogSink
requestID string
startedAt time.Time
accessKeyID uint
protocol protocol.Protocol
clientModel string
usageApplicable bool
attempts []telemetry.Attempt
attemptPricing []frozenAttemptPricing
pendingPricing frozenAttemptPricing
pricingPending bool
outcome requestOutcome
usage telemetry.UsageObservation
now func() time.Time
redactor *redact.Redactor
emitted bool
sink telemetry.RequestLogSink
requestID string
startedAt time.Time
accessKeyID uint
protocol protocol.Protocol
clientModel string
usageApplicable bool
usageDiagnostics usage.Diagnostics
attempts []telemetry.Attempt
attemptPricing []frozenAttemptPricing
pendingPricing frozenAttemptPricing
pricingPending bool
outcome requestOutcome
usage telemetry.UsageObservation
now func() time.Time
redactor *redact.Redactor
emitted bool
pendingRetry int
}
@@ -137,6 +138,12 @@ func (recorder *requestRecorder) setUsageApplicable(applicable bool) {
}
}
func (recorder *requestRecorder) setUsageDiagnostics(diagnostics usage.Diagnostics) {
if recorder != nil {
recorder.usageDiagnostics.Merge(diagnostics)
}
}
func (recorder *requestRecorder) beforeForward() time.Time {
if recorder == nil || recorder.now == nil {
return time.Time{}
@@ -381,6 +388,8 @@ func (recorder *requestRecorder) bindUsage(
result = usage.Result{State: usage.StateNotApplicable}
} else if !validCapturedUsage(result) {
result = usage.Result{State: usage.StateMissing}
} else {
result.Diagnostics.Merge(recorder.usageDiagnostics)
}
frozen := frozenAttemptPricing{}
if attemptIndex < len(recorder.attemptPricing) {
+81
View File
@@ -454,6 +454,48 @@ func TestRequestRecorderPreservesKnownCostWhenUsedOutputPriceIsUnavailable(t *te
}
}
func TestRequestRecorderMergesRequestPricingDiagnosticsIntoKnownCost(t *testing.T) {
table := mustGatewayPriceTable(t, 2_000_000_000, true)
recorder := newRequestRecorder(
&recordingRequestLogSink{}, "unsupported-mode", time.Unix(100, 0), 1,
protocol.OpenAICompletions, func() time.Time { return time.Unix(101, 0) },
)
diagnostics := usage.Diagnostics{}
diagnostics.Add(usage.DiagnosticUnsupportedBillableDetail)
recorder.setUsageDiagnostics(diagnostics)
model := "gpt-4o"
recorder.freezeNextAttemptPricing(frozenAttemptPricing{
groupID: 1, scopeKey: "group:1", upstreamModel: model, table: table, applicable: true,
})
index := recorder.appendAttempt(
scheduler.Selection{
GroupID: 1, Group: state.GroupView{Name: "group"}, KeyID: 2,
UpstreamModelID: &model,
},
UpstreamResult{StatusCode: http.StatusOK},
telemetry.FailureCategoryOK, telemetry.ActionTerminate, "", "",
time.Unix(100, 0), time.Unix(101, 0),
)
recorder.completeResponse(UpstreamResult{
StatusCode: http.StatusOK,
Usage: usage.Result{
State: usage.StateComplete,
Tokens: usage.Tokens{UncachedInput: 1_000_000, Output: 1_000_000},
},
}, health.Result{}, model, index)
if !recorder.usage.Result.Diagnostics.Has(usage.DiagnosticUnsupportedBillableDetail) {
t.Fatalf("usage diagnostics = %#v, want unsupported billable detail", recorder.usage.Result.Diagnostics)
}
if got := recorder.usage.Pricing; got != (telemetry.PricingObservation{
PriceScopeKey: "group:1", UpstreamModel: model,
CostState: "priced", PricingCompleteness: "partial",
EstimatedCostNanoUSD: 4_000_000_000,
}) {
t.Fatalf("unsupported mode pricing = %#v", got)
}
}
func TestHandlerUsesSelectedProviderModelForPricingInsteadOfAliasOrBodyModel(t *testing.T) {
providerID := "openai"
model := "provider-model"
@@ -508,6 +550,45 @@ func TestHandlerUsesSelectedProviderModelForPricingInsteadOfAliasOrBodyModel(t *
}
}
func TestHandlerMarksExplicitUnsupportedPricingModePartial(t *testing.T) {
table := mustGatewayPriceTable(t, 2_000_000_000, true)
forwarder := &scriptedForwarder{results: []UpstreamResult{{
StatusCode: http.StatusOK, Header: make(http.Header), RequestWritten: true,
Usage: usage.Result{
State: usage.StateComplete,
Tokens: usage.Tokens{
UncachedInput: 1_000_000,
Output: 1_000_000,
},
},
}}}
sink := &recordingRequestLogSink{}
engine, handler, _, _ := newRequestLogHandlerTestRuntime(
t, forwarder, &recordingAccessKeyRPMLimiter{}, sink, "sk-first",
)
handler.priceTables = &mutableGatewayPriceTableProvider{table: table}
request := httptest.NewRequest(
http.MethodPost,
"/v1/chat/completions",
strings.NewReader(`{"model":"gpt-4o","service_tier":"priority"}`),
)
request.Header.Set("Authorization", "Bearer gl-client")
engine.ServeHTTP(httptest.NewRecorder(), request)
events := sink.snapshot()
if len(events) != 1 {
t.Fatalf("events = %#v, want one", events)
}
if !events[0].Usage.Result.Diagnostics.Has(usage.DiagnosticUnsupportedBillableDetail) ||
events[0].Usage.Pricing != (telemetry.PricingObservation{
PriceScopeKey: "group:1", UpstreamModel: "gpt-4o",
CostState: "priced", PricingCompleteness: "partial",
EstimatedCostNanoUSD: 4_000_000_000,
}) {
t.Fatalf("unsupported mode event = %#v", events[0])
}
}
func mustGatewayPriceTable(t *testing.T, input int64, outputSet bool) *pricing.Table {
t.Helper()
prices := pricing.Prices{
+5
View File
@@ -629,6 +629,11 @@ func TestAutoMigrateCreatesRequestLogFieldsAndCompositeIndexes(t *testing.T) {
{name: "completed_at_ms", desc: 1},
{name: "id", desc: 1},
},
"idx_request_logs_upstream_model_completed_id": {
{name: "upstream_model"},
{name: "completed_at_ms", desc: 1},
{name: "id", desc: 1},
},
}
type indexedColumn struct {
Sequence int `gorm:"column:seqno"`
+3 -3
View File
@@ -2,13 +2,13 @@ package models
// RequestLog is the durable request-level audit and usage record.
type RequestLog struct {
ID string `gorm:"type:varchar(36);primaryKey;not null;index:idx_request_logs_completed_id,priority:2,sort:desc;index:idx_request_logs_access_completed_id,priority:3,sort:desc;index:idx_request_logs_status_completed_id,priority:3,sort:desc;index:idx_request_logs_model_completed_id,priority:3,sort:desc"`
CompletedAtMS int64 `gorm:"column:completed_at_ms;not null;index:idx_request_logs_completed_id,priority:1,sort:desc;index:idx_request_logs_access_completed_id,priority:2,sort:desc;index:idx_request_logs_status_completed_id,priority:2,sort:desc;index:idx_request_logs_model_completed_id,priority:2,sort:desc"`
ID string `gorm:"type:varchar(36);primaryKey;not null;index:idx_request_logs_completed_id,priority:2,sort:desc;index:idx_request_logs_access_completed_id,priority:3,sort:desc;index:idx_request_logs_status_completed_id,priority:3,sort:desc;index:idx_request_logs_model_completed_id,priority:3,sort:desc;index:idx_request_logs_upstream_model_completed_id,priority:3,sort:desc"`
CompletedAtMS int64 `gorm:"column:completed_at_ms;not null;index:idx_request_logs_completed_id,priority:1,sort:desc;index:idx_request_logs_access_completed_id,priority:2,sort:desc;index:idx_request_logs_status_completed_id,priority:2,sort:desc;index:idx_request_logs_model_completed_id,priority:2,sort:desc;index:idx_request_logs_upstream_model_completed_id,priority:2,sort:desc"`
AccessKeyID uint `gorm:"not null;index:idx_request_logs_access_completed_id,priority:1"`
GroupID uint `gorm:"not null;default:0"`
Protocol string `gorm:"type:varchar(32);not null"`
ClientModel string `gorm:"type:varchar(255);not null;index:idx_request_logs_model_completed_id,priority:1"`
UpstreamModel string `gorm:"type:varchar(255);not null"`
UpstreamModel string `gorm:"type:varchar(255);not null;index:idx_request_logs_upstream_model_completed_id,priority:1"`
Status string `gorm:"type:varchar(32);not null;index:idx_request_logs_status_completed_id,priority:1"`
StatusCode int `gorm:"not null"`
DurationMs int64 `gorm:"not null"`
+2
View File
@@ -252,6 +252,8 @@ func schemaV4IndexStatements() []string {
ON request_logs(status, completed_at_ms DESC, id DESC)`,
`CREATE INDEX idx_request_logs_model_completed_id
ON request_logs(client_model, completed_at_ms DESC, id DESC)`,
`CREATE INDEX idx_request_logs_upstream_model_completed_id
ON request_logs(upstream_model, completed_at_ms DESC, id DESC)`,
`CREATE UNIQUE INDEX idx_usage_stats_bucket_access_group_model
ON usage_stats(bucket_start_ms, access_key_id, group_id, model)`,
`CREATE UNIQUE INDEX idx_model_prices_scope_model
+8 -7
View File
@@ -126,13 +126,14 @@ export default {
},
},
reset: {
open: 'Restore automatic prices',
title: 'Restore automatic prices?',
description: 'Remove the manual prices for “{model}” and use catalog prices again',
close: 'Close automatic-price confirmation',
warning: 'The model returns to pending when the catalog has no matching price',
confirm: 'Restore automatic prices',
failed: 'Unable to restore automatic prices',
open: 'Reset prices',
title: 'Reset prices?',
description: 'Reset the manual price settings for “{model}”',
close: 'Close price reset confirmation',
warning:
'Automatic prices are restored when available; otherwise the model returns to pending',
confirm: 'Reset prices',
failed: 'Unable to reset prices',
},
delete: {
open: 'Delete price entry',
+7 -7
View File
@@ -123,13 +123,13 @@ export default {
},
},
reset: {
open: '自動価格に戻す',
title: '自動価格に戻しますか?',
description: '「{model}」の手動価格を削除し、カタログ価格を再び使用します',
close: '自動価格の確認を閉じる',
warning: 'カタログに対応する価格がない場合、このモデルは価格待ちに戻ります',
confirm: '自動価格に戻す',
failed: '自動価格に戻せません',
open: '価格をリセット',
title: '価格をリセットしますか?',
description: '「{model}」の手動価格設定をリセットします',
close: '価格リセットの確認を閉じる',
warning: '利用可能な場合は自動価格に戻り、ない場合は価格待ちになります',
confirm: '価格をリセット',
failed: '価格をリセットできません',
},
delete: {
open: '価格レコードを削除',
+7 -7
View File
@@ -122,13 +122,13 @@ export default {
},
},
reset: {
open: '恢复自动价格',
title: '恢复自动价格?',
description: '移除“{model}”的手动价格并重新采用目录价格',
close: '关闭恢复自动价格确认',
warning: '目录中没有对应价格时,该模型会回到待定价状态',
confirm: '恢复自动价格',
failed: '无法恢复自动价格',
open: '重置价格',
title: '重置价格?',
description: '重置“{model}”的手动价格设置',
close: '关闭重置价格确认',
warning: '有可用目录价格时恢复自动价格,否则回到待定价状态',
confirm: '重置价格',
failed: '无法重置价格',
},
delete: {
open: '删除价格记录',