diff --git a/internal/catalog/official.go b/internal/catalog/official.go index ef3e673b..5cfe55c2 100644 --- a/internal/catalog/official.go +++ b/internal/catalog/official.go @@ -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 { diff --git a/internal/catalog/official_test.go b/internal/catalog/official_test.go index 84fda137..7eeb6e88 100644 --- a/internal/catalog/official_test.go +++ b/internal/catalog/official_test.go @@ -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"}, diff --git a/internal/control/bootstrap.go b/internal/control/bootstrap.go index d2d0c80d..23d57c02 100644 --- a/internal/control/bootstrap.go +++ b/internal/control/bootstrap.go @@ -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) diff --git a/internal/control/catalog_discovery_merge_test.go b/internal/control/catalog_discovery_merge_test.go index 8607735f..d1783ef9 100644 --- a/internal/control/catalog_discovery_merge_test.go +++ b/internal/control/catalog_discovery_merge_test.go @@ -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) diff --git a/internal/control/catalog_sync.go b/internal/control/catalog_sync.go index 8384a4aa..e5dcbf9e 100644 --- a/internal/control/catalog_sync.go +++ b/internal/control/catalog_sync.go @@ -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 { diff --git a/internal/control/catalog_sync_test.go b/internal/control/catalog_sync_test.go index be691309..bc45b986 100644 --- a/internal/control/catalog_sync_test.go +++ b/internal/control/catalog_sync_test.go @@ -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) diff --git a/internal/control/discover_executor.go b/internal/control/discover_executor.go index 964a7d49..bee598b8 100644 --- a/internal/control/discover_executor.go +++ b/internal/control/discover_executor.go @@ -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 { diff --git a/internal/control/group_create.go b/internal/control/group_create.go index 8f41ed7e..ad446c49 100644 --- a/internal/control/group_create.go +++ b/internal/control/group_create.go @@ -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 } diff --git a/internal/control/group_create_test.go b/internal/control/group_create_test.go index b70d0ffc..428f7833 100644 --- a/internal/control/group_create_test.go +++ b/internal/control/group_create_test.go @@ -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 := " " diff --git a/internal/control/group_settings.go b/internal/control/group_settings.go index 3c050d1a..39c838a9 100644 --- a/internal/control/group_settings.go +++ b/internal/control/group_settings.go @@ -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) diff --git a/internal/control/operation_recovery_test.go b/internal/control/operation_recovery_test.go index 04f1d9a0..393d1fd9 100644 --- a/internal/control/operation_recovery_test.go +++ b/internal/control/operation_recovery_test.go @@ -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) { diff --git a/internal/control/price_reconcile_test.go b/internal/control/price_reconcile_test.go index 440ba267..acf3693b 100644 --- a/internal/control/price_reconcile_test.go +++ b/internal/control/price_reconcile_test.go @@ -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 { diff --git a/internal/control/pricing_status.go b/internal/control/pricing_status.go index 880880d3..48ce9377 100644 --- a/internal/control/pricing_status.go +++ b/internal/control/pricing_status.go @@ -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 +} diff --git a/internal/control/provider_catalog.go b/internal/control/provider_catalog.go index 7fb98a63..9c46a635 100644 --- a/internal/control/provider_catalog.go +++ b/internal/control/provider_catalog.go @@ -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 } diff --git a/internal/control/provider_catalog_http_test.go b/internal/control/provider_catalog_http_test.go index 383a47fa..269a3de3 100644 --- a/internal/control/provider_catalog_http_test.go +++ b/internal/control/provider_catalog_http_test.go @@ -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, }) diff --git a/internal/control/service.go b/internal/control/service.go index 870a0c78..c331a03b 100644 --- a/internal/control/service.go +++ b/internal/control/service.go @@ -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 diff --git a/internal/control/settings.go b/internal/control/settings.go index 54cb2820..aa5f55f7 100644 --- a/internal/control/settings.go +++ b/internal/control/settings.go @@ -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, diff --git a/internal/control/settings_test.go b/internal/control/settings_test.go index a900f765..499e4c16 100644 --- a/internal/control/settings_test.go +++ b/internal/control/settings_test.go @@ -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) diff --git a/internal/dialect/dialect.go b/internal/dialect/dialect.go index f6a11564..a5bf8f79 100644 --- a/internal/dialect/dialect.go +++ b/internal/dialect/dialect.go @@ -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 { diff --git a/internal/dialect/openai.go b/internal/dialect/openai.go index 95adc63c..2e3c6b6a 100644 --- a/internal/dialect/openai.go +++ b/internal/dialect/openai.go @@ -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 } diff --git a/internal/dialect/openai_pricing_modes.go b/internal/dialect/openai_pricing_modes.go new file mode 100644 index 00000000..35504399 --- /dev/null +++ b/internal/dialect/openai_pricing_modes.go @@ -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 +} diff --git a/internal/dialect/openai_responses.go b/internal/dialect/openai_responses.go index ea5df54b..2368ca7f 100644 --- a/internal/dialect/openai_responses.go +++ b/internal/dialect/openai_responses.go @@ -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 } diff --git a/internal/dialect/openai_responses_test.go b/internal/dialect/openai_responses_test.go index d3e81451..b6a4f96f 100644 --- a/internal/dialect/openai_responses_test.go +++ b/internal/dialect/openai_responses_test.go @@ -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() diff --git a/internal/dialect/openai_responses_usage.go b/internal/dialect/openai_responses_usage.go index 36e4e194..b4f6bfb6 100644 --- a/internal/dialect/openai_responses_usage.go +++ b/internal/dialect/openai_responses_usage.go @@ -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) } diff --git a/internal/dialect/openai_responses_usage_test.go b/internal/dialect/openai_responses_usage_test.go index 81fc0069..9744658f 100644 --- a/internal/dialect/openai_responses_usage_test.go +++ b/internal/dialect/openai_responses_usage_test.go @@ -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}}`, diff --git a/internal/dialect/openai_test.go b/internal/dialect/openai_test.go index 4b441c62..bf1bf2c1 100644 --- a/internal/dialect/openai_test.go +++ b/internal/dialect/openai_test.go @@ -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 { diff --git a/internal/dialect/openai_usage.go b/internal/dialect/openai_usage.go index 78308ee7..f5ffb402 100644 --- a/internal/dialect/openai_usage.go +++ b/internal/dialect/openai_usage.go @@ -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) } diff --git a/internal/dialect/openai_usage_test.go b/internal/dialect/openai_usage_test.go index 7c2013f9..8dd29a80 100644 --- a/internal/dialect/openai_usage_test.go +++ b/internal/dialect/openai_usage_test.go @@ -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}}}`, diff --git a/internal/dialect/usage_json.go b/internal/dialect/usage_json.go index e0f12454..6d619fe6 100644 --- a/internal/dialect/usage_json.go +++ b/internal/dialect/usage_json.go @@ -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 diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go index 9ae38901..4d493d04 100644 --- a/internal/gateway/handler.go +++ b/internal/gateway/handler.go @@ -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 { diff --git a/internal/gateway/request_log.go b/internal/gateway/request_log.go index 5f31fb2e..8d6de32c 100644 --- a/internal/gateway/request_log.go +++ b/internal/gateway/request_log.go @@ -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) { diff --git a/internal/gateway/request_log_test.go b/internal/gateway/request_log_test.go index 20d38a94..a84e3f64 100644 --- a/internal/gateway/request_log_test.go +++ b/internal/gateway/request_log_test.go @@ -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{ diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go index 5a115c84..74c210c3 100644 --- a/internal/storage/db_test.go +++ b/internal/storage/db_test.go @@ -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"` diff --git a/internal/storage/models/request.go b/internal/storage/models/request.go index 27792432..8f19a6c3 100644 --- a/internal/storage/models/request.go +++ b/internal/storage/models/request.go @@ -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"` diff --git a/internal/storage/schema_v4.go b/internal/storage/schema_v4.go index e7cd5c19..1f78c5d2 100644 --- a/internal/storage/schema_v4.go +++ b/internal/storage/schema_v4.go @@ -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 diff --git a/web/src/i18n/locales/en-US/model-prices.ts b/web/src/i18n/locales/en-US/model-prices.ts index 2701999a..62693341 100644 --- a/web/src/i18n/locales/en-US/model-prices.ts +++ b/web/src/i18n/locales/en-US/model-prices.ts @@ -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', diff --git a/web/src/i18n/locales/ja-JP/model-prices.ts b/web/src/i18n/locales/ja-JP/model-prices.ts index 69fbb56c..2b053484 100644 --- a/web/src/i18n/locales/ja-JP/model-prices.ts +++ b/web/src/i18n/locales/ja-JP/model-prices.ts @@ -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: '価格レコードを削除', diff --git a/web/src/i18n/locales/zh-CN/model-prices.ts b/web/src/i18n/locales/zh-CN/model-prices.ts index eae8640d..8e9ea720 100644 --- a/web/src/i18n/locales/zh-CN/model-prices.ts +++ b/web/src/i18n/locales/zh-CN/model-prices.ts @@ -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: '删除价格记录',