diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index ba5b78a72f..75786b9c70 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -196,7 +196,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { accountUsageService := service.NewAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService) accountTestService := service.NewAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService) crsSyncService := service.NewCRSSyncService(accountRepository, proxyRepository, oAuthService, openAIOAuthService, geminiOAuthService, configConfig) - accountHandler := admin.NewAccountHandler(adminService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, rateLimitService, accountUsageService, accountTestService, concurrencyService, crsSyncService, sessionLimitCache, rpmCache, compositeTokenCacheInvalidator) + accountHandler := admin.ProvideAccountHandler(adminService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, rateLimitService, accountUsageService, accountTestService, concurrencyService, crsSyncService, sessionLimitCache, rpmCache, compositeTokenCacheInvalidator, grokQuotaService) adminAnnouncementHandler := admin.NewAnnouncementHandler(announcementService) dataManagementService := service.NewDataManagementService() dataManagementHandler := admin.NewDataManagementHandler(dataManagementService) diff --git a/backend/ent/channelmonitor/channelmonitor.go b/backend/ent/channelmonitor/channelmonitor.go index afdc6957d6..711e6217e2 100644 --- a/backend/ent/channelmonitor/channelmonitor.go +++ b/backend/ent/channelmonitor/channelmonitor.go @@ -167,6 +167,7 @@ const ( ProviderOpenai Provider = "openai" ProviderAnthropic Provider = "anthropic" ProviderGemini Provider = "gemini" + ProviderGrok Provider = "grok" ) func (pr Provider) String() string { @@ -176,7 +177,7 @@ func (pr Provider) String() string { // ProviderValidator is a validator for the "provider" field enum values. It is called by the builders before save. func ProviderValidator(pr Provider) error { switch pr { - case ProviderOpenai, ProviderAnthropic, ProviderGemini: + case ProviderOpenai, ProviderAnthropic, ProviderGemini, ProviderGrok: return nil default: return fmt.Errorf("channelmonitor: invalid enum value for provider field: %q", pr) diff --git a/backend/ent/channelmonitorrequesttemplate/channelmonitorrequesttemplate.go b/backend/ent/channelmonitorrequesttemplate/channelmonitorrequesttemplate.go index db04aee106..5989d0e743 100644 --- a/backend/ent/channelmonitorrequesttemplate/channelmonitorrequesttemplate.go +++ b/backend/ent/channelmonitorrequesttemplate/channelmonitorrequesttemplate.go @@ -103,6 +103,7 @@ const ( ProviderOpenai Provider = "openai" ProviderAnthropic Provider = "anthropic" ProviderGemini Provider = "gemini" + ProviderGrok Provider = "grok" ) func (pr Provider) String() string { @@ -112,7 +113,7 @@ func (pr Provider) String() string { // ProviderValidator is a validator for the "provider" field enum values. It is called by the builders before save. func ProviderValidator(pr Provider) error { switch pr { - case ProviderOpenai, ProviderAnthropic, ProviderGemini: + case ProviderOpenai, ProviderAnthropic, ProviderGemini, ProviderGrok: return nil default: return fmt.Errorf("channelmonitorrequesttemplate: invalid enum value for provider field: %q", pr) diff --git a/backend/ent/migrate/schema.go b/backend/ent/migrate/schema.go index cac559d535..52229f9151 100644 --- a/backend/ent/migrate/schema.go +++ b/backend/ent/migrate/schema.go @@ -623,7 +623,7 @@ var ( {Name: "created_at", Type: field.TypeTime, SchemaType: map[string]string{"postgres": "timestamptz"}}, {Name: "updated_at", Type: field.TypeTime, SchemaType: map[string]string{"postgres": "timestamptz"}}, {Name: "name", Type: field.TypeString, Size: 100}, - {Name: "provider", Type: field.TypeEnum, Enums: []string{"openai", "anthropic", "gemini"}}, + {Name: "provider", Type: field.TypeEnum, Enums: []string{"openai", "anthropic", "gemini", "grok"}}, {Name: "api_mode", Type: field.TypeString, Size: 32, Default: "chat_completions"}, {Name: "endpoint", Type: field.TypeString, Size: 500}, {Name: "api_key_encrypted", Type: field.TypeString}, @@ -768,7 +768,7 @@ var ( {Name: "created_at", Type: field.TypeTime, SchemaType: map[string]string{"postgres": "timestamptz"}}, {Name: "updated_at", Type: field.TypeTime, SchemaType: map[string]string{"postgres": "timestamptz"}}, {Name: "name", Type: field.TypeString, Size: 100}, - {Name: "provider", Type: field.TypeEnum, Enums: []string{"openai", "anthropic", "gemini"}}, + {Name: "provider", Type: field.TypeEnum, Enums: []string{"openai", "anthropic", "gemini", "grok"}}, {Name: "api_mode", Type: field.TypeString, Size: 32, Default: "chat_completions"}, {Name: "description", Type: field.TypeString, Nullable: true, Size: 500, Default: ""}, {Name: "extra_headers", Type: field.TypeJSON}, diff --git a/backend/ent/schema/channel_monitor.go b/backend/ent/schema/channel_monitor.go index d9594ab39c..cb62079316 100644 --- a/backend/ent/schema/channel_monitor.go +++ b/backend/ent/schema/channel_monitor.go @@ -35,7 +35,7 @@ func (ChannelMonitor) Fields() []ent.Field { NotEmpty(). MaxLen(100), field.Enum("provider"). - Values("openai", "anthropic", "gemini"), + Values("openai", "anthropic", "gemini", "grok"), field.String("api_mode"). Default("chat_completions"). MaxLen(32). diff --git a/backend/ent/schema/channel_monitor_request_template.go b/backend/ent/schema/channel_monitor_request_template.go index 0e0ce3a0b5..cf7fe05158 100644 --- a/backend/ent/schema/channel_monitor_request_template.go +++ b/backend/ent/schema/channel_monitor_request_template.go @@ -39,7 +39,7 @@ func (ChannelMonitorRequestTemplate) Fields() []ent.Field { NotEmpty(). MaxLen(100), field.Enum("provider"). - Values("openai", "anthropic", "gemini"), + Values("openai", "anthropic", "gemini", "grok"), field.String("api_mode"). Default("chat_completions"). MaxLen(32). diff --git a/backend/internal/handler/admin/account_data.go b/backend/internal/handler/admin/account_data.go index bf872c4826..e44d726fd6 100644 --- a/backend/internal/handler/admin/account_data.go +++ b/backend/internal/handler/admin/account_data.go @@ -460,6 +460,7 @@ func (h *AccountHandler) importData(ctx context.Context, req DataImportRequest) if created.Platform == service.PlatformAntigravity && created.Type == service.AccountTypeOAuth { privacyAccounts = append(privacyAccounts, created) } + h.scheduleGrokImportProbe(created) result.AccountCreated++ } diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index b886728159..e4ed5b46b0 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -60,6 +60,7 @@ type AccountHandler struct { sessionLimitCache service.SessionLimitCache rpmCache service.RPMCache tokenCacheInvalidator service.TokenCacheInvalidator + grokImportProber grokUsageProber } // NewAccountHandler creates a new admin account handler @@ -855,6 +856,7 @@ func (h *AccountHandler) Create(c *gin.Context) { // OpenAI APIKey 账号创建后异步探测上游 /v1/responses 能力。 // 探测失败不影响账号创建响应。 h.scheduleOpenAIResponsesProbe(createdAccount) + h.scheduleGrokImportProbe(createdAccount) response.Success(c, result.Data) } @@ -1667,6 +1669,7 @@ func (h *AccountHandler) BatchCreate(c *gin.Context) { } // OpenAI APIKey 账号异步探测 /v1/responses 能力。 h.scheduleOpenAIResponsesProbe(account) + h.scheduleGrokImportProbe(account) success++ results = append(results, gin.H{ "name": item.Name, diff --git a/backend/internal/handler/admin/channel_monitor_handler.go b/backend/internal/handler/admin/channel_monitor_handler.go index 4ef774e9e7..a69b835849 100644 --- a/backend/internal/handler/admin/channel_monitor_handler.go +++ b/backend/internal/handler/admin/channel_monitor_handler.go @@ -37,11 +37,11 @@ func NewChannelMonitorHandler(monitorService *service.ChannelMonitorService) *Ch type channelMonitorCreateRequest struct { Name string `json:"name" binding:"required,max=100"` - Provider string `json:"provider" binding:"required,oneof=openai anthropic gemini"` + Provider string `json:"provider" binding:"required,oneof=openai anthropic gemini grok"` APIMode string `json:"api_mode" binding:"omitempty,oneof=chat_completions responses"` Endpoint string `json:"endpoint" binding:"required,max=500"` APIKey string `json:"api_key" binding:"required,max=2000"` - PrimaryModel string `json:"primary_model" binding:"required,max=200"` + PrimaryModel string `json:"primary_model" binding:"max=200"` ExtraModels []string `json:"extra_models"` GroupName string `json:"group_name" binding:"max=100"` Enabled *bool `json:"enabled"` @@ -55,7 +55,7 @@ type channelMonitorCreateRequest struct { type channelMonitorUpdateRequest struct { Name *string `json:"name" binding:"omitempty,max=100"` - Provider *string `json:"provider" binding:"omitempty,oneof=openai anthropic gemini"` + Provider *string `json:"provider" binding:"omitempty,oneof=openai anthropic gemini grok"` APIMode *string `json:"api_mode" binding:"omitempty,oneof=chat_completions responses"` Endpoint *string `json:"endpoint" binding:"omitempty,max=500"` APIKey *string `json:"api_key" binding:"omitempty,max=2000"` diff --git a/backend/internal/handler/admin/channel_monitor_template_handler.go b/backend/internal/handler/admin/channel_monitor_template_handler.go index c842f465c8..497e3d195b 100644 --- a/backend/internal/handler/admin/channel_monitor_template_handler.go +++ b/backend/internal/handler/admin/channel_monitor_template_handler.go @@ -26,7 +26,7 @@ func NewChannelMonitorRequestTemplateHandler(templateService *service.ChannelMon type channelMonitorTemplateCreateRequest struct { Name string `json:"name" binding:"required,max=100"` - Provider string `json:"provider" binding:"required,oneof=openai anthropic gemini"` + Provider string `json:"provider" binding:"required,oneof=openai anthropic gemini grok"` APIMode string `json:"api_mode" binding:"omitempty,oneof=chat_completions responses"` Description string `json:"description" binding:"max=500"` ExtraHeaders map[string]string `json:"extra_headers"` diff --git a/backend/internal/handler/admin/grok_import_probe.go b/backend/internal/handler/admin/grok_import_probe.go new file mode 100644 index 0000000000..f1df15bba9 --- /dev/null +++ b/backend/internal/handler/admin/grok_import_probe.go @@ -0,0 +1,203 @@ +package admin + +import ( + "context" + "log/slog" + "sync" + "time" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/Wei-Shaw/sub2api/internal/service" +) + +const ( + grokImportProbeConcurrency = 3 + grokImportProbeTimeout = 25 * time.Second +) + +type grokUsageProber interface { + ProbeUsage(ctx context.Context, accountID int64) (*service.GrokQuotaProbeResult, error) +} + +type grokImportProbeTask struct { + prober grokUsageProber + accountID int64 +} + +type grokImportProbeScheduler struct { + mu sync.Mutex + queue []grokImportProbeTask + concurrency int + workers int + maxWorkers int + timeout time.Duration +} + +var defaultGrokImportProbeScheduler = newGrokImportProbeScheduler( + grokImportProbeConcurrency, + grokImportProbeTimeout, +) + +func newGrokImportProbeScheduler(concurrency int, timeout time.Duration) *grokImportProbeScheduler { + if concurrency <= 0 { + concurrency = 1 + } + if timeout <= 0 { + timeout = grokImportProbeTimeout + } + return &grokImportProbeScheduler{ + concurrency: concurrency, + timeout: timeout, + } +} + +func (s *grokImportProbeScheduler) schedule(prober grokUsageProber, account *service.Account) { + if s == nil || prober == nil || account == nil || account.ID <= 0 { + return + } + if account.Platform != service.PlatformGrok || account.Type != service.AccountTypeOAuth { + return + } + + s.mu.Lock() + s.queue = append(s.queue, grokImportProbeTask{prober: prober, accountID: account.ID}) + if s.workers < s.concurrency { + s.workers++ + if s.workers > s.maxWorkers { + s.maxWorkers = s.workers + } + go s.worker() + } + s.mu.Unlock() +} + +func (s *grokImportProbeScheduler) worker() { + for { + task, ok := s.nextTask() + if !ok { + return + } + s.run(task.prober, task.accountID) + } +} + +func (s *grokImportProbeScheduler) nextTask() (grokImportProbeTask, bool) { + s.mu.Lock() + defer s.mu.Unlock() + if len(s.queue) == 0 { + s.workers-- + return grokImportProbeTask{}, false + } + task := s.queue[0] + s.queue[0] = grokImportProbeTask{} + s.queue = s.queue[1:] + if len(s.queue) == 0 { + s.queue = nil + } + return task, true +} + +func (s *grokImportProbeScheduler) run(prober grokUsageProber, accountID int64) { + defer func() { + if recovered := recover(); recovered != nil { + slog.Error( + "grok_import_active_probe_panic", + "account_id", accountID, + "recovery_type", panicType(recovered), + ) + } + }() + + // Queue time is intentionally excluded: every imported account is probed, + // while this timeout only bounds the actual upstream probe execution. + ctx, cancel := context.WithTimeout(context.Background(), s.timeout) + defer cancel() + result, err := prober.ProbeUsage(ctx, accountID) + if err != nil { + slog.Warn( + "grok_import_active_probe_failed", + "account_id", accountID, + "status", infraerrors.Code(err), + "reason", infraerrors.Reason(err), + ) + return + } + if result == nil { + slog.Warn( + "grok_import_active_probe_failed", + "account_id", accountID, + "reason", "empty_result", + ) + return + } + + slog.Info( + "grok_import_active_probe_completed", + "account_id", accountID, + "model", result.Model, + "status", result.StatusCode, + "headers_observed", result.HeadersObserved, + ) +} + +func panicType(value any) string { + switch value.(type) { + case string: + return "string" + case error: + return "error" + default: + return "unknown" + } +} + +func (h *AccountHandler) scheduleGrokImportProbe(account *service.Account) { + if h == nil { + return + } + defaultGrokImportProbeScheduler.schedule(h.grokImportProber, account) +} + +func (h *GrokOAuthHandler) scheduleGrokImportProbe(account *service.Account) { + if h == nil { + return + } + defaultGrokImportProbeScheduler.schedule(h.importProber, account) +} + +// ProvideAccountHandler injects the Grok active prober for production while +// keeping NewAccountHandler convenient for focused unit tests. +func ProvideAccountHandler( + adminService service.AdminService, + oauthService *service.OAuthService, + openaiOAuthService *service.OpenAIOAuthService, + geminiOAuthService *service.GeminiOAuthService, + antigravityOAuthService *service.AntigravityOAuthService, + rateLimitService *service.RateLimitService, + accountUsageService *service.AccountUsageService, + accountTestService *service.AccountTestService, + concurrencyService *service.ConcurrencyService, + crsSyncService *service.CRSSyncService, + sessionLimitCache service.SessionLimitCache, + rpmCache service.RPMCache, + tokenCacheInvalidator service.TokenCacheInvalidator, + grokQuotaService *service.GrokQuotaService, +) *AccountHandler { + handler := NewAccountHandler( + adminService, + oauthService, + openaiOAuthService, + geminiOAuthService, + antigravityOAuthService, + rateLimitService, + accountUsageService, + accountTestService, + concurrencyService, + crsSyncService, + sessionLimitCache, + rpmCache, + tokenCacheInvalidator, + ) + handler.grokImportProber = grokQuotaService + return handler +} diff --git a/backend/internal/handler/admin/grok_import_probe_handler_test.go b/backend/internal/handler/admin/grok_import_probe_handler_test.go new file mode 100644 index 0000000000..489a13ae6d --- /dev/null +++ b/backend/internal/handler/admin/grok_import_probe_handler_test.go @@ -0,0 +1,119 @@ +//go:build unit + +package admin + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/Wei-Shaw/sub2api/internal/pkg/xai" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +type grokImportAdminService struct { + *stubAdminService + mu sync.Mutex + nextID int64 +} + +func newGrokImportAdminService() *grokImportAdminService { + return &grokImportAdminService{ + stubAdminService: newStubAdminService(), + nextID: 500, + } +} + +func (s *grokImportAdminService) CreateAccount(_ context.Context, input *service.CreateAccountInput) (*service.Account, error) { + s.mu.Lock() + s.nextID++ + id := s.nextID + s.mu.Unlock() + return &service.Account{ + ID: id, + Name: input.Name, + Platform: input.Platform, + Type: input.Type, + Credentials: input.Credentials, + Extra: input.Extra, + ProxyID: input.ProxyID, + Concurrency: input.Concurrency, + Status: service.StatusActive, + Schedulable: true, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + }, nil +} + +type grokImportOAuthClientStub struct{} + +func (grokImportOAuthClientStub) ExchangeCode(context.Context, string, string, string, string, string) (*xai.TokenResponse, error) { + return &xai.TokenResponse{AccessToken: "access-token", RefreshToken: "refresh-token", ExpiresIn: 3600}, nil +} + +func (grokImportOAuthClientStub) RefreshToken(context.Context, string, string, string) (*xai.TokenResponse, error) { + return &xai.TokenResponse{AccessToken: "access-token", RefreshToken: "refresh-token", ExpiresIn: 3600}, nil +} + +func (grokImportOAuthClientStub) ConvertSSOToBuild(context.Context, string, string) (*xai.TokenResponse, error) { + return &xai.TokenResponse{AccessToken: "access-token", RefreshToken: "refresh-token", ExpiresIn: 3600}, nil +} + +func TestGrokSSOBatchImportKeepsCreatedAccountsWhenOneAutomaticProbeFails(t *testing.T) { + gin.SetMode(gin.TestMode) + adminService := newGrokImportAdminService() + oauthService := service.NewGrokOAuthService(nil, grokImportOAuthClientStub{}) + defer oauthService.Stop() + prober := newGrokImportProbeStub(3) + prober.failures[502] = infraerrors.New(502, "GROK_TEST_PROBE_FAILED", "sensitive-upstream-body") + handler := NewGrokOAuthHandler(oauthService, adminService, nil) + handler.importProber = prober + + router := gin.New() + router.POST("/api/v1/admin/grok/sso-to-oauth", handler.CreateAccountsFromSSO) + recorder := httptest.NewRecorder() + request := httptest.NewRequest( + http.MethodPost, + "/api/v1/admin/grok/sso-to-oauth", + strings.NewReader(`{"sso_tokens":["sso-one","sso-two","sso-three"]}`), + ) + request.Header.Set("Content-Type", "application/json") + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Contains(t, recorder.Body.String(), `"created"`) + require.NotContains(t, recorder.Body.String(), `GROK_TEST_PROBE_FAILED`) + for i := 0; i < 3; i++ { + awaitGrokProbeSignal(t, prober.done) + } + calls, _, _ := prober.snapshot() + require.Equal(t, map[int64]int{501: 1, 502: 1, 503: 1}, calls) +} + +func TestAccountCreateWithoutAutomaticGrokProbeServiceStillSucceeds(t *testing.T) { + gin.SetMode(gin.TestMode) + handler := NewAccountHandler( + newGrokImportAdminService(), + nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, + ) + + router := gin.New() + router.POST("/api/v1/admin/accounts", handler.Create) + recorder := httptest.NewRecorder() + request := httptest.NewRequest( + http.MethodPost, + "/api/v1/admin/accounts", + strings.NewReader(`{"name":"grok-rt","platform":"grok","type":"oauth","credentials":{"refresh_token":"secret"}}`), + ) + request.Header.Set("Content-Type", "application/json") + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) +} diff --git a/backend/internal/handler/admin/grok_import_probe_test.go b/backend/internal/handler/admin/grok_import_probe_test.go new file mode 100644 index 0000000000..3b8fc0ca6e --- /dev/null +++ b/backend/internal/handler/admin/grok_import_probe_test.go @@ -0,0 +1,232 @@ +//go:build unit + +package admin + +import ( + "bytes" + "context" + "log/slog" + "sync" + "testing" + "time" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +type grokImportProbeStub struct { + mu sync.Mutex + calls map[int64]int + failures map[int64]error + active int + maxActive int + deadlineSeen bool + block <-chan struct{} + started chan int64 + done chan int64 +} + +func newGrokImportProbeStub(buffer int) *grokImportProbeStub { + return &grokImportProbeStub{ + calls: make(map[int64]int), + failures: make(map[int64]error), + started: make(chan int64, buffer), + done: make(chan int64, buffer), + } +} + +func (s *grokImportProbeStub) ProbeUsage(ctx context.Context, accountID int64) (*service.GrokQuotaProbeResult, error) { + _, deadlineSeen := ctx.Deadline() + s.mu.Lock() + s.calls[accountID]++ + s.active++ + if s.active > s.maxActive { + s.maxActive = s.active + } + s.deadlineSeen = s.deadlineSeen || deadlineSeen + s.mu.Unlock() + + s.started <- accountID + var ctxErr error + if s.block != nil { + select { + case <-s.block: + case <-ctx.Done(): + ctxErr = ctx.Err() + } + } + + s.mu.Lock() + s.active-- + failure := s.failures[accountID] + s.mu.Unlock() + s.done <- accountID + if ctxErr != nil { + return nil, ctxErr + } + if failure != nil { + return nil, failure + } + return &service.GrokQuotaProbeResult{ + Source: "active_probe", + Model: "grok-4.5", + StatusCode: 200, + ResetSupported: false, + }, nil +} + +func (s *grokImportProbeStub) snapshot() (map[int64]int, int, bool) { + s.mu.Lock() + defer s.mu.Unlock() + calls := make(map[int64]int, len(s.calls)) + for id, count := range s.calls { + calls[id] = count + } + return calls, s.maxActive, s.deadlineSeen +} + +type grokImportProbeSchedulerTestSnapshot struct { + queued int + workers int + maxWorkers int +} + +func snapshotGrokImportProbeScheduler(s *grokImportProbeScheduler) grokImportProbeSchedulerTestSnapshot { + if s == nil { + return grokImportProbeSchedulerTestSnapshot{} + } + s.mu.Lock() + defer s.mu.Unlock() + return grokImportProbeSchedulerTestSnapshot{ + queued: len(s.queue), + workers: s.workers, + maxWorkers: s.maxWorkers, + } +} + +func newGrokOAuthImportAccount(id int64) *service.Account { + return &service.Account{ + ID: id, + Platform: service.PlatformGrok, + Type: service.AccountTypeOAuth, + } +} + +func awaitGrokProbeSignal(t *testing.T, signals <-chan int64) int64 { + t.Helper() + select { + case id := <-signals: + return id + case <-time.After(time.Second): + t.Fatal("timed out waiting for Grok import probe") + return 0 + } +} + +func TestGrokImportProbeSchedulerProbesSingleAccountOnce(t *testing.T) { + scheduler := newGrokImportProbeScheduler(1, time.Second) + prober := newGrokImportProbeStub(1) + + scheduler.schedule(prober, newGrokOAuthImportAccount(101)) + require.Equal(t, int64(101), awaitGrokProbeSignal(t, prober.done)) + + calls, maxActive, deadlineSeen := prober.snapshot() + require.Equal(t, map[int64]int{101: 1}, calls) + require.Equal(t, 1, maxActive) + require.True(t, deadlineSeen) + require.Eventually(t, func() bool { + snapshot := snapshotGrokImportProbeScheduler(scheduler) + return snapshot.queued == 0 && snapshot.workers == 0 + }, time.Second, 10*time.Millisecond) +} + +func TestGrokImportProbeSchedulerQueuesBatchWithoutPerTaskGoroutines(t *testing.T) { + const taskCount = 100 + release := make(chan struct{}) + scheduler := newGrokImportProbeScheduler(3, time.Second) + prober := newGrokImportProbeStub(taskCount) + prober.block = release + prober.failures[150] = infraerrors.New(502, "GROK_TEST_PROBE_FAILED", "sensitive-upstream-body") + + for id := int64(101); id < 101+taskCount; id++ { + scheduler.schedule(prober, newGrokOAuthImportAccount(id)) + } + for i := 0; i < 3; i++ { + awaitGrokProbeSignal(t, prober.started) + } + snapshot := snapshotGrokImportProbeScheduler(scheduler) + require.Equal(t, 97, snapshot.queued) + require.Equal(t, 3, snapshot.workers) + require.Equal(t, 3, snapshot.maxWorkers) + select { + case id := <-prober.started: + t.Fatalf("probe %d started before a concurrency slot was released", id) + case <-time.After(75 * time.Millisecond): + } + close(release) + for i := 0; i < taskCount; i++ { + awaitGrokProbeSignal(t, prober.done) + } + + calls, maxActive, _ := prober.snapshot() + require.Len(t, calls, taskCount) + for id := int64(101); id < 101+taskCount; id++ { + require.Equal(t, 1, calls[id]) + } + require.Equal(t, 3, maxActive) + require.Eventually(t, func() bool { + snapshot = snapshotGrokImportProbeScheduler(scheduler) + return snapshot.queued == 0 && snapshot.workers == 0 + }, time.Second, 10*time.Millisecond) + require.Equal(t, 3, snapshot.maxWorkers) +} + +func TestGrokImportProbeSchedulerTimeoutCancelsProbe(t *testing.T) { + neverRelease := make(chan struct{}) + scheduler := newGrokImportProbeScheduler(1, 20*time.Millisecond) + prober := newGrokImportProbeStub(1) + prober.block = neverRelease + + scheduler.schedule(prober, newGrokOAuthImportAccount(201)) + require.Equal(t, int64(201), awaitGrokProbeSignal(t, prober.done)) + + calls, _, _ := prober.snapshot() + require.Equal(t, 1, calls[201]) +} + +func TestGrokImportProbeSchedulerSkipsMissingServiceAndNonGrokAccounts(t *testing.T) { + scheduler := newGrokImportProbeScheduler(1, time.Second) + prober := newGrokImportProbeStub(1) + + scheduler.schedule(nil, newGrokOAuthImportAccount(301)) + scheduler.schedule(prober, &service.Account{ID: 302, Platform: service.PlatformOpenAI, Type: service.AccountTypeOAuth}) + scheduler.schedule(prober, &service.Account{ID: 303, Platform: service.PlatformGrok, Type: service.AccountTypeAPIKey}) + + select { + case id := <-prober.started: + t.Fatalf("unexpected probe for account %d", id) + case <-time.After(50 * time.Millisecond): + } + calls, _, _ := prober.snapshot() + require.Empty(t, calls) +} + +func TestGrokImportProbeFailureLogDoesNotIncludeErrorMessage(t *testing.T) { + var logs bytes.Buffer + previousLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + defer slog.SetDefault(previousLogger) + + scheduler := newGrokImportProbeScheduler(1, time.Second) + prober := newGrokImportProbeStub(1) + prober.failures[401] = infraerrors.New(502, "GROK_TEST_PROBE_FAILED", "refresh-token-secret") + scheduler.schedule(prober, newGrokOAuthImportAccount(401)) + awaitGrokProbeSignal(t, prober.done) + + require.Eventually(t, func() bool { + return bytes.Contains(logs.Bytes(), []byte("grok_import_active_probe_failed")) + }, time.Second, 10*time.Millisecond) + require.Contains(t, logs.String(), "GROK_TEST_PROBE_FAILED") + require.NotContains(t, logs.String(), "refresh-token-secret") +} diff --git a/backend/internal/handler/admin/grok_oauth_handler.go b/backend/internal/handler/admin/grok_oauth_handler.go index dfe5632e30..1a309b7c9a 100644 --- a/backend/internal/handler/admin/grok_oauth_handler.go +++ b/backend/internal/handler/admin/grok_oauth_handler.go @@ -22,6 +22,7 @@ type GrokOAuthHandler struct { grokOAuthService *service.GrokOAuthService adminService service.AdminService quotaService *service.GrokQuotaService + importProber grokUsageProber } func NewGrokOAuthHandler( @@ -33,6 +34,7 @@ func NewGrokOAuthHandler( grokOAuthService: grokOAuthService, adminService: adminService, quotaService: quotaService, + importProber: quotaService, } } @@ -209,6 +211,7 @@ func (h *GrokOAuthHandler) CreateAccountFromOAuth(c *gin.Context) { response.ErrorFrom(c, err) return } + h.scheduleGrokImportProbe(account) response.Success(c, dto.AccountFromService(account)) } @@ -345,6 +348,7 @@ func (h *GrokOAuthHandler) createAccountFromSSOToken(ctx context.Context, req Gr if err != nil { return grokSSOImportWorkerResult{item: GrokSSOToOAuthItemResult{Index: index, Name: name, Email: tokenInfo.Email, Error: grokSSOImportErrorMessage(err)}} } + h.scheduleGrokImportProbe(account) return grokSSOImportWorkerResult{ created: true, item: GrokSSOToOAuthItemResult{ diff --git a/backend/internal/handler/wire.go b/backend/internal/handler/wire.go index cfbb72554c..67380fed33 100644 --- a/backend/internal/handler/wire.go +++ b/backend/internal/handler/wire.go @@ -164,7 +164,7 @@ var ProviderSet = wire.NewSet( admin.NewDashboardHandler, admin.NewUserHandler, admin.NewGroupHandler, - admin.NewAccountHandler, + admin.ProvideAccountHandler, admin.NewAnnouncementHandler, admin.NewDataManagementHandler, admin.NewBackupHandler, diff --git a/backend/internal/service/channel_monitor_checker.go b/backend/internal/service/channel_monitor_checker.go index 889b2bbed7..ad4058f9e6 100644 --- a/backend/internal/service/channel_monitor_checker.go +++ b/backend/internal/service/channel_monitor_checker.go @@ -168,6 +168,7 @@ type providerAdapter struct { //nolint:gochecknoglobals // 适配器表是只读静态数据，初始化后不变更。 var providerAdapters = map[string]providerAdapter{ MonitorProviderOpenAI: providerOpenAIChatAdapter, + MonitorProviderGrok: providerGrokChatAdapter, MonitorProviderAnthropic: { buildPath: func(string) string { return providerAnthropicPath }, buildBody: func(model, prompt string) ([]byte, error) { @@ -205,20 +206,27 @@ var providerAdapters = map[string]providerAdapter{ } //nolint:gochecknoglobals // 适配器表是只读静态数据，初始化后不变更。 -var providerOpenAIChatAdapter = providerAdapter{ - buildPath: func(string) string { return providerOpenAIPath }, - buildBody: func(model, prompt string) ([]byte, error) { - return json.Marshal(map[string]any{ - "model": model, - "messages": []map[string]string{{"role": "user", "content": prompt}}, - "max_tokens": monitorChallengeMaxTokens, - "stream": false, - }) - }, - buildHeaders: func(apiKey string) map[string]string { - return map[string]string{"Authorization": "Bearer " + apiKey} - }, - textPath: "choices.0.message.content", +var providerOpenAIChatAdapter = newOpenAICompatibleChatAdapter(providerOpenAIPath) + +//nolint:gochecknoglobals // 适配器表是只读静态数据，初始化后不变更。 +var providerGrokChatAdapter = newOpenAICompatibleChatAdapter(providerGrokPath) + +func newOpenAICompatibleChatAdapter(path string) providerAdapter { + return providerAdapter{ + buildPath: func(string) string { return path }, + buildBody: func(model, prompt string) ([]byte, error) { + return json.Marshal(map[string]any{ + "model": model, + "messages": []map[string]string{{"role": "user", "content": prompt}}, + "max_tokens": monitorChallengeMaxTokens, + "stream": false, + }) + }, + buildHeaders: func(apiKey string) map[string]string { + return map[string]string{"Authorization": "Bearer " + apiKey} + }, + textPath: "choices.0.message.content", + } } //nolint:gochecknoglobals // 适配器表是只读静态数据，初始化后不变更。 @@ -408,8 +416,9 @@ func buildRequestBody(adapter providerAdapter, provider, apiMode, model, prompt var bodyMergeKeyDenyList = map[string]map[string]bool{ MonitorProviderOpenAI + ":" + MonitorAPIModeChatCompletions: {"model": true, "messages": true, "stream": true}, MonitorProviderOpenAI + ":" + MonitorAPIModeResponses: {"model": true, "instructions": true, "input": true, "stream": true}, - MonitorProviderAnthropic: {"model": true, "messages": true}, - MonitorProviderGemini: {"contents": true}, + MonitorProviderGrok: {"model": true, "messages": true, "stream": true}, + MonitorProviderAnthropic: {"model": true, "messages": true}, + MonitorProviderGemini: {"contents": true}, } func checkAPIMode(opts *CheckOptions) string { @@ -427,7 +436,7 @@ func bodyMergeDenyKey(provider, apiMode string) string { } func validateReplaceRequestBody(provider, apiMode string, body map[string]any) error { - if provider != MonitorProviderOpenAI { + if provider != MonitorProviderOpenAI && provider != MonitorProviderGrok { return nil } switch defaultAPIMode(apiMode) { @@ -528,6 +537,8 @@ var monitorAPIKeyPatterns = []struct { {regexp.MustCompile(`sk-ant-[A-Za-z0-9_-]{20,}`), "sk-ant-***REDACTED***"}, // OpenAI / Anthropic 通用 sk-: sk-xxxxxxx {regexp.MustCompile(`sk-[A-Za-z0-9-]{20,}`), "sk-***REDACTED***"}, + // xAI API Key：xai-xxxxxxx + {regexp.MustCompile(`xai-[A-Za-z0-9_-]{6,}`), "xai-***REDACTED***"}, // Gemini / Google API Key：固定前缀 + 35 位 {regexp.MustCompile(`AIza[A-Za-z0-9_-]{35}`), "AIza***REDACTED***"}, // JWT 三段式（Bearer 后常出现）：eyJxxx.eyJxxx.signature @@ -537,7 +548,7 @@ var monitorAPIKeyPatterns = []struct { // sanitizeErrorMessage 擦除错误/响应文本中可能泄露的 API key。 // 处理两类来源： // 1. URL query 中的 ?key= / ?api_key= 等（Go *url.Error 会回填完整 URL） -// 2. 上游 HTTP body 文本里直接出现的 sk-* / AIza* / JWT 等密钥碎片 +// 2. 上游 HTTP body 文本里直接出现的 sk-* / xai-* / AIza* / JWT 等密钥碎片 // // 注意：与 gemini_messages_compat_service.go 的 sanitizeUpstreamErrorMessage 关注点类似但参数集更广， // 监控模块独立维护，避免互相耦合。 diff --git a/backend/internal/service/channel_monitor_checker_body_test.go b/backend/internal/service/channel_monitor_checker_body_test.go index bba3d7dfb7..bcf7af0b98 100644 --- a/backend/internal/service/channel_monitor_checker_body_test.go +++ b/backend/internal/service/channel_monitor_checker_body_test.go @@ -64,6 +64,7 @@ type openAICaptureHandler struct { lastHeaders http.Header lastPath string status int + rawResponse string responsesLeadingReasoning bool } @@ -80,6 +81,10 @@ func (h *openAICaptureHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) } w.Header().Set("Content-Type", "application/json") w.WriteHeader(h.status) + if h.rawResponse != "" { + _, _ = w.Write([]byte(h.rawResponse)) + return + } answer := answerFromOpenAIRequest(parsed) if h.lastPath == providerOpenAIResponsesPath { @@ -190,6 +195,90 @@ func TestRunCheckForModel_OpenAI_DefaultChatRequest(t *testing.T) { } } +func TestGrokMonitorConfiguration(t *testing.T) { + if err := validateProvider(MonitorProviderGrok); err != nil { + t.Fatalf("grok provider should be supported: %v", err) + } + if got := normalizeMonitorPrimaryModel(MonitorProviderGrok, ""); got != MonitorDefaultGrokModel { + t.Fatalf("expected default Grok model %q, got %q", MonitorDefaultGrokModel, got) + } + if err := validateAPIMode(MonitorProviderGrok, MonitorAPIModeChatCompletions); err != nil { + t.Fatalf("grok chat_completions mode should be valid: %v", err) + } + if err := validateAPIMode(MonitorProviderGrok, MonitorAPIModeResponses); err == nil { + t.Fatal("grok responses mode should be rejected by channel monitoring") + } + if err := validateReplaceRequestBody(MonitorProviderGrok, MonitorAPIModeChatCompletions, map[string]any{}); err == nil { + t.Fatal("grok replace-mode body should require messages") + } +} + +func TestRunCheckForModel_Grok_DefaultChatRequest(t *testing.T) { + h := &openAICaptureHandler{} + endpoint := setupFakeOpenAI(t, h) + + res := runCheckForModel(context.Background(), MonitorProviderGrok, endpoint, "xai-key", MonitorDefaultGrokModel, nil) + + if res.Status != MonitorStatusOperational { + t.Fatalf("Grok request should pass challenge, got status=%s message=%q", res.Status, res.Message) + } + if res.LatencyMs == nil { + t.Fatal("Grok request should record latency") + } + if h.lastPath != providerGrokPath { + t.Fatalf("expected Grok chat completions path %q, got %q", providerGrokPath, h.lastPath) + } + if h.lastBody["model"] != MonitorDefaultGrokModel { + t.Errorf("Grok body should contain model=%s, got %v", MonitorDefaultGrokModel, h.lastBody["model"]) + } + if _, ok := h.lastBody["messages"]; !ok { + t.Error("Grok body should contain messages") + } + if h.lastBody["stream"] != false { + t.Errorf("Grok body should set stream=false, got %v", h.lastBody["stream"]) + } + if h.lastHeaders.Get("Authorization") != "Bearer xai-key" { + t.Errorf("expected Grok bearer auth header, got %q", h.lastHeaders.Get("Authorization")) + } +} + +func TestRunCheckForModel_Grok_UpstreamFailure(t *testing.T) { + h := &openAICaptureHandler{status: http.StatusTooManyRequests} + endpoint := setupFakeOpenAI(t, h) + + res := runCheckForModel(context.Background(), MonitorProviderGrok, endpoint, "xai-key", MonitorDefaultGrokModel, nil) + + if res.Status != MonitorStatusError { + t.Fatalf("Grok 429 should be recorded as error, got status=%s message=%q", res.Status, res.Message) + } + if !strings.Contains(res.Message, "upstream HTTP 429") { + t.Fatalf("Grok failure should preserve upstream status, got %q", res.Message) + } + if res.LatencyMs == nil { + t.Fatal("Grok failure should still record latency") + } +} + +func TestRunCheckForModel_Grok_RedactsXAIKeyFromUpstreamBody(t *testing.T) { + h := &openAICaptureHandler{ + status: http.StatusUnauthorized, + rawResponse: `{"error":{"message":"invalid API key xai-secret"}}`, + } + endpoint := setupFakeOpenAI(t, h) + + res := runCheckForModel(context.Background(), MonitorProviderGrok, endpoint, "request-key", MonitorDefaultGrokModel, nil) + + if res.Status != MonitorStatusError { + t.Fatalf("Grok upstream failure should be recorded as error, got %s", res.Status) + } + if strings.Contains(res.Message, "xai-secret") { + t.Fatalf("Grok error message leaked xAI key: %q", res.Message) + } + if !strings.Contains(res.Message, "xai-***REDACTED***") { + t.Fatalf("Grok error message should contain redaction marker, got %q", res.Message) + } +} + func TestRunCheckForModel_OpenAIResponses_DefaultRequest(t *testing.T) { h := &openAICaptureHandler{} endpoint := setupFakeOpenAI(t, h) diff --git a/backend/internal/service/channel_monitor_const.go b/backend/internal/service/channel_monitor_const.go index 61f9f79894..2ee8eabde8 100644 --- a/backend/internal/service/channel_monitor_const.go +++ b/backend/internal/service/channel_monitor_const.go @@ -47,6 +47,8 @@ const ( // providerOpenAIPath OpenAI Chat Completions 路径。 providerOpenAIPath = "/v1/chat/completions" + // providerGrokPath Grok OpenAI-compatible Chat Completions 路径。 + providerGrokPath = "/v1/chat/completions" // providerOpenAIResponsesPath OpenAI Responses API 路径。 providerOpenAIResponsesPath = "/v1/responses" // providerAnthropicPath Anthropic Messages 路径。 @@ -54,10 +56,14 @@ const ( // providerGeminiPathTemplate Gemini generateContent 路径模板（含 model 占位）。 providerGeminiPathTemplate = "/v1beta/models/%s:generateContent" - // MonitorProviderOpenAI / Anthropic / Gemini provider 字符串常量（也是 ent enum 的实际值）。 + // MonitorProviderOpenAI / Anthropic / Gemini / Grok provider 字符串常量（也是 ent enum 的实际值）。 MonitorProviderOpenAI = "openai" MonitorProviderAnthropic = "anthropic" MonitorProviderGemini = "gemini" + MonitorProviderGrok = "grok" + + // MonitorDefaultGrokModel 是新增 Grok 监控未显式指定模型时使用的轻量测活模型。 + MonitorDefaultGrokModel = "grok-4.5" // MonitorStatusOperational 等监控状态字符串常量（与 ent enum 一致）。 MonitorStatusOperational = "operational" @@ -112,13 +118,13 @@ var ( "CHANNEL_MONITOR_NOT_FOUND", "channel monitor not found", ) ErrChannelMonitorInvalidProvider = infraerrors.BadRequest( - "CHANNEL_MONITOR_INVALID_PROVIDER", "provider must be one of openai/anthropic/gemini", + "CHANNEL_MONITOR_INVALID_PROVIDER", "provider must be one of openai/anthropic/gemini/grok", ) ErrChannelMonitorInvalidAPIMode = infraerrors.BadRequest( "CHANNEL_MONITOR_INVALID_API_MODE", "api_mode must be chat_completions or responses; responses is only supported for openai", ) ErrChannelMonitorInvalidRequestBody = infraerrors.BadRequest( - "CHANNEL_MONITOR_INVALID_REQUEST_BODY", "openai replace-mode body_override must include non-empty messages for chat_completions or non-empty instructions and input for responses", + "CHANNEL_MONITOR_INVALID_REQUEST_BODY", "openai-compatible replace-mode body_override must include non-empty messages for chat_completions or non-empty instructions and input for responses", ) ErrChannelMonitorInvalidInterval = infraerrors.BadRequest( "CHANNEL_MONITOR_INVALID_INTERVAL", "interval_seconds must be in [15, 3600]", diff --git a/backend/internal/service/channel_monitor_service.go b/backend/internal/service/channel_monitor_service.go index 7b53bb20b0..b5dea22589 100644 --- a/backend/internal/service/channel_monitor_service.go +++ b/backend/internal/service/channel_monitor_service.go @@ -123,7 +123,7 @@ func (s *ChannelMonitorService) Create(ctx context.Context, p ChannelMonitorCrea APIMode: defaultAPIMode(p.APIMode), Endpoint: normalizeEndpoint(p.Endpoint), APIKey: encrypted, // 注意：传入 repository 时该字段为密文 - PrimaryModel: strings.TrimSpace(p.PrimaryModel), + PrimaryModel: normalizeMonitorPrimaryModel(p.Provider, p.PrimaryModel), ExtraModels: normalizeModels(p.ExtraModels), GroupName: strings.TrimSpace(p.GroupName), Enabled: p.Enabled, @@ -167,7 +167,7 @@ func validateCreateParams(p ChannelMonitorCreateParams) error { if strings.TrimSpace(p.APIKey) == "" { return ErrChannelMonitorMissingAPIKey } - if strings.TrimSpace(p.PrimaryModel) == "" { + if normalizeMonitorPrimaryModel(p.Provider, p.PrimaryModel) == "" { return ErrChannelMonitorMissingPrimaryModel } return nil @@ -486,8 +486,8 @@ func applyMonitorUpdate(existing *ChannelMonitor, p ChannelMonitorUpdateParams) if err := validateProvider(*p.Provider); err != nil { return err } + providerChanged = existing.Provider != *p.Provider existing.Provider = *p.Provider - providerChanged = true } if p.Endpoint != nil { if err := validateEndpoint(*p.Endpoint); err != nil { @@ -496,7 +496,13 @@ func applyMonitorUpdate(existing *ChannelMonitor, p ChannelMonitorUpdateParams) existing.Endpoint = normalizeEndpoint(*p.Endpoint) } if p.PrimaryModel != nil { - existing.PrimaryModel = strings.TrimSpace(*p.PrimaryModel) + primaryModel := normalizeMonitorPrimaryModel(existing.Provider, *p.PrimaryModel) + if primaryModel == "" { + return ErrChannelMonitorMissingPrimaryModel + } + existing.PrimaryModel = primaryModel + } else if providerChanged && existing.Provider == MonitorProviderGrok { + existing.PrimaryModel = MonitorDefaultGrokModel } if p.ExtraModels != nil { existing.ExtraModels = normalizeModels(*p.ExtraModels) diff --git a/backend/internal/service/channel_monitor_service_grok_test.go b/backend/internal/service/channel_monitor_service_grok_test.go new file mode 100644 index 0000000000..20c9db2666 --- /dev/null +++ b/backend/internal/service/channel_monitor_service_grok_test.go @@ -0,0 +1,85 @@ +//go:build unit + +package service + +import "testing" + +func TestApplyMonitorUpdate_ProviderOnlySwitchToGrokUsesDefaultModel(t *testing.T) { + grok := MonitorProviderGrok + existing := &ChannelMonitor{ + Provider: MonitorProviderOpenAI, + APIMode: MonitorAPIModeResponses, + PrimaryModel: "gpt-5", + IntervalSeconds: 60, + } + + err := applyMonitorUpdate(existing, ChannelMonitorUpdateParams{Provider: &grok}) + if err != nil { + t.Fatalf("provider-only switch to Grok failed: %v", err) + } + if existing.PrimaryModel != MonitorDefaultGrokModel { + t.Fatalf("expected Grok default model %q, got %q", MonitorDefaultGrokModel, existing.PrimaryModel) + } + if existing.APIMode != MonitorAPIModeChatCompletions { + t.Fatalf("expected Grok API mode %q, got %q", MonitorAPIModeChatCompletions, existing.APIMode) + } +} + +func TestApplyMonitorUpdate_SwitchToGrokPreservesExplicitModel(t *testing.T) { + grok := MonitorProviderGrok + explicitModel := "grok-4.3" + existing := &ChannelMonitor{ + Provider: MonitorProviderOpenAI, + APIMode: MonitorAPIModeChatCompletions, + PrimaryModel: "gpt-5", + IntervalSeconds: 60, + } + + err := applyMonitorUpdate(existing, ChannelMonitorUpdateParams{ + Provider: &grok, + PrimaryModel: &explicitModel, + }) + if err != nil { + t.Fatalf("switch to Grok with explicit model failed: %v", err) + } + if existing.PrimaryModel != explicitModel { + t.Fatalf("expected explicit model %q, got %q", explicitModel, existing.PrimaryModel) + } +} + +func TestApplyMonitorUpdate_SameGrokProviderDoesNotResetExistingModel(t *testing.T) { + grok := MonitorProviderGrok + existing := &ChannelMonitor{ + Provider: MonitorProviderGrok, + APIMode: MonitorAPIModeChatCompletions, + PrimaryModel: "grok-4.3", + IntervalSeconds: 60, + } + + err := applyMonitorUpdate(existing, ChannelMonitorUpdateParams{Provider: &grok}) + if err != nil { + t.Fatalf("same-provider Grok update failed: %v", err) + } + if existing.PrimaryModel != "grok-4.3" { + t.Fatalf("same-provider update reset existing model to %q", existing.PrimaryModel) + } +} + +func TestApplyMonitorUpdate_SwitchToGrokRejectsResponsesMode(t *testing.T) { + grok := MonitorProviderGrok + responses := MonitorAPIModeResponses + existing := &ChannelMonitor{ + Provider: MonitorProviderOpenAI, + APIMode: MonitorAPIModeChatCompletions, + PrimaryModel: "gpt-5", + IntervalSeconds: 60, + } + + err := applyMonitorUpdate(existing, ChannelMonitorUpdateParams{ + Provider: &grok, + APIMode: &responses, + }) + if err == nil { + t.Fatal("Grok responses mode should remain unsupported") + } +} diff --git a/backend/internal/service/channel_monitor_template_types.go b/backend/internal/service/channel_monitor_template_types.go index 03cd518d28..0b824d577d 100644 --- a/backend/internal/service/channel_monitor_template_types.go +++ b/backend/internal/service/channel_monitor_template_types.go @@ -55,7 +55,7 @@ var ( "CHANNEL_MONITOR_TEMPLATE_NOT_FOUND", "channel monitor request template not found", ) ErrChannelMonitorTemplateInvalidProvider = infraerrors.BadRequest( - "CHANNEL_MONITOR_TEMPLATE_INVALID_PROVIDER", "template provider must be one of openai/anthropic/gemini", + "CHANNEL_MONITOR_TEMPLATE_INVALID_PROVIDER", "template provider must be one of openai/anthropic/gemini/grok", ) ErrChannelMonitorTemplateInvalidAPIMode = infraerrors.BadRequest( "CHANNEL_MONITOR_TEMPLATE_INVALID_API_MODE", "template api_mode must be chat_completions or responses; responses is only supported for openai", diff --git a/backend/internal/service/channel_monitor_validate.go b/backend/internal/service/channel_monitor_validate.go index c5a4783b91..7740dc83b7 100644 --- a/backend/internal/service/channel_monitor_validate.go +++ b/backend/internal/service/channel_monitor_validate.go @@ -124,6 +124,16 @@ func normalizeModels(in []string) []string { return out } +// normalizeMonitorPrimaryModel applies the Grok health-check default while +// preserving the existing required-model behavior for every other provider. +func normalizeMonitorPrimaryModel(provider, model string) string { + model = strings.TrimSpace(model) + if model == "" && provider == MonitorProviderGrok { + return MonitorDefaultGrokModel + } + return model +} + // defaultAPIMode 空串归一为 chat_completions，保证历史数据与旧客户端兼容。 func defaultAPIMode(apiMode string) string { if strings.TrimSpace(apiMode) == "" { diff --git a/backend/internal/service/grok_quota_service.go b/backend/internal/service/grok_quota_service.go index 7817a00ec4..2219a75c15 100644 --- a/backend/internal/service/grok_quota_service.go +++ b/backend/internal/service/grok_quota_service.go @@ -183,10 +183,22 @@ func (s *GrokQuotaService) probeUsage(ctx context.Context, accountID int64) (*Gr return result, nil } if resp.StatusCode >= 400 { - bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, 240)) - bodyText := truncate(strings.TrimSpace(string(bodyBytes)), 240) - slog.Warn("grok_quota_probe_failed", "account_id", account.ID, "model", probeModel, "status", resp.StatusCode, "body", bodyText) - return nil, infraerrors.Newf(mapUpstreamStatus(resp.StatusCode), "GROK_QUOTA_PROBE_UPSTREAM_ERROR", "upstream returned %d for probe model %q: %s", resp.StatusCode, probeModel, bodyText) + const reason = "GROK_QUOTA_PROBE_UPSTREAM_ERROR" + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) + slog.Warn( + "grok_quota_probe_failed", + "account_id", account.ID, + "model", probeModel, + "status", resp.StatusCode, + "reason", reason, + ) + return nil, infraerrors.Newf( + mapUpstreamStatus(resp.StatusCode), + reason, + "upstream returned %d for probe model %q", + resp.StatusCode, + probeModel, + ) } return result, nil } diff --git a/backend/internal/service/grok_quota_service_test.go b/backend/internal/service/grok_quota_service_test.go index 7a72f80e9f..ba9b6cbcb8 100644 --- a/backend/internal/service/grok_quota_service_test.go +++ b/backend/internal/service/grok_quota_service_test.go @@ -3,8 +3,10 @@ package service import ( + "bytes" "context" "io" + "log/slog" "net/http" "strconv" "strings" @@ -298,6 +300,54 @@ func TestGrokQuotaServiceProbeUsageReportsProbeModelOnUpstreamError(t *testing.T require.Contains(t, infraerrors.Message(err), `probe model "grok-4.5"`) } +func TestGrokQuotaServiceProbeUsageRedactsUpstreamErrorBodyFromErrorAndLogs(t *testing.T) { + const upstreamSecret = "upstream-secret-refresh-token" + account := &Account{ + ID: 49, + Platform: PlatformGrok, + Type: AccountTypeOAuth, + Concurrency: 1, + Credentials: map[string]any{ + "access_token": "access-token", + "expires_at": time.Now().Add(time.Hour).UTC().Format(time.RFC3339), + }, + } + repo := &grokQuotaAccountRepo{ + mockAccountRepoForPlatform: &mockAccountRepoForPlatform{ + accountsByID: map[int64]*Account{49: account}, + }, + } + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusBadRequest, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader( + `{"error":"` + upstreamSecret + `","detail":"credential rejected"}`, + )), + }} + svc := NewGrokQuotaService( + repo, + nil, + NewGrokTokenProvider(repo, nil), + upstream, + ) + + var logs bytes.Buffer + previousLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + defer slog.SetDefault(previousLogger) + + _, err := svc.ProbeUsage(context.Background(), account.ID) + require.Error(t, err) + require.Equal(t, "GROK_QUOTA_PROBE_UPSTREAM_ERROR", infraerrors.Reason(err)) + require.Contains(t, infraerrors.Message(err), `probe model "grok-4.5"`) + require.NotContains(t, err.Error(), upstreamSecret) + require.NotContains(t, infraerrors.Message(err), upstreamSecret) + require.Contains(t, logs.String(), "GROK_QUOTA_PROBE_UPSTREAM_ERROR") + require.NotContains(t, logs.String(), upstreamSecret) + require.NotContains(t, logs.String(), "credential rejected") + require.Equal(t, xai.DefaultCLIBaseURL+"/responses", upstream.lastReq.URL.String()) +} + func TestGrokQuotaServiceProbeUsageLoadsProxyWhenAccountEdgeMissing(t *testing.T) { t.Parallel() diff --git a/backend/internal/service/openai_content_session_seed.go b/backend/internal/service/openai_content_session_seed.go index 7c2ba25140..fce85f11bd 100644 --- a/backend/internal/service/openai_content_session_seed.go +++ b/backend/internal/service/openai_content_session_seed.go @@ -11,6 +11,10 @@ import ( // and explicit session IDs (e.g. "sess-xxx" or "compat_cc_xxx"). const contentSessionSeedPrefix = "compat_cs_" +// contentStablePrefixSessionSeedPrefix distinguishes cache identities derived +// only from request fields that remain stable across independent prompts. +const contentStablePrefixSessionSeedPrefix = "compat_csp_" + // deriveOpenAIContentSessionSeed builds a stable session seed from an // OpenAI-format request body. Only fields constant across conversation turns // are included: model, tools/functions definitions, system/developer prompts, @@ -105,3 +109,156 @@ func deriveOpenAIContentSessionSeed(body []byte) string { } return contentSessionSeedPrefix + b.String() } + +// deriveOpenAIAnchoredContentSessionSeed returns the legacy content-derived +// seed only when it contains a meaningful user/input anchor. This preserves +// the existing session derivation while preventing model-only requests from +// becoming a tenant-wide cache routing identity. +func deriveOpenAIAnchoredContentSessionSeed(body []byte) string { + if !hasOpenAIContentSessionUserAnchor(body) { + return "" + } + return deriveOpenAIContentSessionSeed(body) +} + +func hasOpenAIContentSessionUserAnchor(body []byte) bool { + if len(body) == 0 { + return false + } + + if messages := gjson.GetBytes(body, "messages"); messages.Exists() && messages.IsArray() { + anchored := false + messages.ForEach(func(_, message gjson.Result) bool { + if strings.TrimSpace(message.Get("role").String()) != "user" { + return true + } + anchored = hasMeaningfulOpenAIContent(message.Get("content")) + return false + }) + return anchored + } + + input := gjson.GetBytes(body, "input") + if !input.Exists() { + return false + } + if input.Type == gjson.String { + return strings.TrimSpace(input.String()) != "" + } + if !input.IsArray() { + return false + } + + anchored := false + input.ForEach(func(_, item gjson.Result) bool { + if strings.TrimSpace(item.Get("role").String()) == "user" { + anchored = hasMeaningfulOpenAIContent(item.Get("content")) + return false + } + if strings.TrimSpace(item.Get("type").String()) == "input_text" { + anchored = strings.TrimSpace(item.Get("text").String()) != "" + return false + } + return true + }) + return anchored +} + +func hasMeaningfulOpenAIContent(content gjson.Result) bool { + if !content.Exists() || content.Type == gjson.Null { + return false + } + if content.Type == gjson.String { + return strings.TrimSpace(content.String()) != "" + } + if !content.IsArray() { + normalized, ok := normalizeNonEmptyCompatSeedJSON(content) + return ok && strings.TrimSpace(normalized) != "" + } + + meaningful := false + content.ForEach(func(_, item gjson.Result) bool { + if item.Type == gjson.String { + meaningful = strings.TrimSpace(item.String()) != "" + } else if text := item.Get("text"); text.Exists() { + meaningful = strings.TrimSpace(text.String()) != "" + } else { + _, meaningful = normalizeNonEmptyCompatSeedJSON(item) + } + return !meaningful + }) + return meaningful +} + +// deriveOpenAIStablePrefixSessionSeed builds a seed from the reusable prefix +// of an OpenAI-format request. User and assistant content are deliberately +// excluded so independent prompts with the same system/tool prefix can share +// an upstream prompt-cache routing identity. +// +// An empty result means the request has no meaningful stable prefix. Callers +// must then use a narrower fallback instead of grouping all requests by tenant +// and model alone. +func deriveOpenAIStablePrefixSessionSeed(body []byte) string { + if len(body) == 0 { + return "" + } + + var b strings.Builder + hasStablePrefix := false + appendJSON := func(label string, value gjson.Result) { + normalized, ok := normalizeNonEmptyCompatSeedJSON(value) + if !ok { + return + } + _, _ = b.WriteString("|") + _, _ = b.WriteString(label) + _, _ = b.WriteString("=") + _, _ = b.WriteString(normalized) + hasStablePrefix = true + } + + if tools := gjson.GetBytes(body, "tools"); tools.Exists() && tools.IsArray() { + appendJSON("tools", tools) + } + if funcs := gjson.GetBytes(body, "functions"); funcs.Exists() && funcs.IsArray() { + appendJSON("functions", funcs) + } + if instructions := gjson.GetBytes(body, "instructions"); strings.TrimSpace(instructions.String()) != "" { + appendJSON("instructions", instructions) + } + + appendSystemMessages := func(items gjson.Result) { + items.ForEach(func(_, item gjson.Result) bool { + role := strings.TrimSpace(item.Get("role").String()) + switch role { + case "system", "developer": + appendJSON(role, item.Get("content")) + } + return true + }) + } + + if messages := gjson.GetBytes(body, "messages"); messages.Exists() && messages.IsArray() { + appendSystemMessages(messages) + } else if input := gjson.GetBytes(body, "input"); input.Exists() && input.IsArray() { + appendSystemMessages(input) + } + + if !hasStablePrefix { + return "" + } + return contentStablePrefixSessionSeedPrefix + b.String() +} + +func normalizeNonEmptyCompatSeedJSON(value gjson.Result) (string, bool) { + if !value.Exists() || value.Type == gjson.Null { + return "", false + } + normalized := normalizeCompatSeedJSON(json.RawMessage(value.Raw)) + switch normalized { + case "", `""`, "[]", "{}", "null": + return "", false + default: + return normalized, true + } +} diff --git a/backend/internal/service/openai_content_session_seed_test.go b/backend/internal/service/openai_content_session_seed_test.go index 65a0bf1808..6dadc5cf53 100644 --- a/backend/internal/service/openai_content_session_seed_test.go +++ b/backend/internal/service/openai_content_session_seed_test.go @@ -216,3 +216,154 @@ func TestDeriveOpenAIContentSessionSeed_ResponsesAPI_TypedMessageItem(t *testing require.Contains(t, seed, "|first_user=") require.Contains(t, seed, "Hello from typed message") } + +func TestDeriveOpenAIStablePrefixSessionSeed_IgnoresUserContent(t *testing.T) { + first := []byte(`{ + "model": "grok", + "instructions": "Be concise.", + "tools": [{"type":"function","name":"lookup","parameters":{"type":"object"}}], + "input": [{"role":"user","content":"Question A"}] + }`) + second := []byte(`{ + "model": "grok", + "instructions": "Be concise.", + "tools": [{"parameters":{"type":"object"},"name":"lookup","type":"function"}], + "input": [{"role":"user","content":"Question B"}] + }`) + + firstSeed := deriveOpenAIStablePrefixSessionSeed(first) + secondSeed := deriveOpenAIStablePrefixSessionSeed(second) + + require.NotEmpty(t, firstSeed) + require.Equal(t, firstSeed, secondSeed) + require.NotContains(t, firstSeed, "Question A") + require.NotContains(t, firstSeed, "first_user") +} + +func TestDeriveOpenAIStablePrefixSessionSeed_IsolatesStablePrefixFields(t *testing.T) { + base := []byte(`{ + "instructions":"Be concise.", + "tools":[{"type":"function","name":"lookup"}], + "input":[{"role":"system","content":"System A"},{"role":"user","content":"Question"}] + }`) + differentInstructions := []byte(`{ + "instructions":"Be detailed.", + "tools":[{"type":"function","name":"lookup"}], + "input":[{"role":"system","content":"System A"},{"role":"user","content":"Question"}] + }`) + differentTools := []byte(`{ + "instructions":"Be concise.", + "tools":[{"type":"function","name":"search"}], + "input":[{"role":"system","content":"System A"},{"role":"user","content":"Question"}] + }`) + differentSystem := []byte(`{ + "instructions":"Be concise.", + "tools":[{"type":"function","name":"lookup"}], + "input":[{"role":"system","content":"System B"},{"role":"user","content":"Question"}] + }`) + + baseSeed := deriveOpenAIStablePrefixSessionSeed(base) + require.NotEqual(t, baseSeed, deriveOpenAIStablePrefixSessionSeed(differentInstructions)) + require.NotEqual(t, baseSeed, deriveOpenAIStablePrefixSessionSeed(differentTools)) + require.NotEqual(t, baseSeed, deriveOpenAIStablePrefixSessionSeed(differentSystem)) +} + +func TestDeriveOpenAIStablePrefixSessionSeed_ChatSystemAndDeveloper(t *testing.T) { + first := []byte(`{ + "messages":[ + {"role":"system","content":"System prompt"}, + {"role":"developer","content":[{"type":"text","text":"Developer prompt"}]}, + {"role":"user","content":"Question A"} + ] + }`) + second := []byte(`{ + "messages":[ + {"role":"system","content":"System prompt"}, + {"role":"developer","content":[{"text":"Developer prompt","type":"text"}]}, + {"role":"user","content":"Question B"} + ] + }`) + + firstSeed := deriveOpenAIStablePrefixSessionSeed(first) + require.Equal(t, firstSeed, deriveOpenAIStablePrefixSessionSeed(second)) + require.Contains(t, firstSeed, "System prompt") + require.Contains(t, firstSeed, "Developer prompt") +} + +func TestDeriveOpenAIStablePrefixSessionSeed_EncodesSystemAndDeveloperRoles(t *testing.T) { + systemThenDeveloper := []byte(`{ + "messages":[ + {"role":"system","content":"Prompt A"}, + {"role":"developer","content":"Prompt B"} + ] + }`) + developerThenSystem := []byte(`{ + "messages":[ + {"role":"developer","content":"Prompt A"}, + {"role":"system","content":"Prompt B"} + ] + }`) + + firstSeed := deriveOpenAIStablePrefixSessionSeed(systemThenDeveloper) + secondSeed := deriveOpenAIStablePrefixSessionSeed(developerThenSystem) + + require.NotEqual(t, firstSeed, secondSeed) + require.Contains(t, firstSeed, "|system=") + require.Contains(t, firstSeed, "|developer=") +} + +func TestDeriveOpenAIStablePrefixSessionSeed_EncodesInstructionDelimiters(t *testing.T) { + instructionOnly := []byte(`{ + "instructions":"foo|system=\"bar\"" + }`) + instructionAndSystem := []byte(`{ + "instructions":"foo", + "input":[{"role":"system","content":"bar"}] + }`) + + firstSeed := deriveOpenAIStablePrefixSessionSeed(instructionOnly) + secondSeed := deriveOpenAIStablePrefixSessionSeed(instructionAndSystem) + + require.NotEmpty(t, firstSeed) + require.NotEmpty(t, secondSeed) + require.NotEqual(t, firstSeed, secondSeed) +} + +func TestDeriveOpenAIAnchoredContentSessionSeed_RequiresMeaningfulAnchor(t *testing.T) { + emptyAnchors := [][]byte{ + nil, + []byte(`{"model":"grok"}`), + []byte(`{"model":"grok","messages":[{"role":"assistant","content":"answer"}]}`), + []byte(`{"model":"grok","messages":[{"role":"user","content":" "}]}`), + []byte(`{"model":"grok","messages":[{"role":"user","content":[{"type":"text","text":""}]}]}`), + []byte(`{"model":"grok","input":" "}`), + []byte(`{"model":"grok","input":[{"type":"input_text","text":""}]}`), + } + for _, body := range emptyAnchors { + require.Empty(t, deriveOpenAIAnchoredContentSessionSeed(body)) + } + + meaningfulAnchors := [][]byte{ + []byte(`{"model":"grok","messages":[{"role":"user","content":"question"}]}`), + []byte(`{"model":"grok","messages":[{"role":"user","content":[{"type":"text","text":"question"}]}]}`), + []byte(`{"model":"grok","input":"question"}`), + []byte(`{"model":"grok","input":[{"type":"input_text","text":"question"}]}`), + } + for _, body := range meaningfulAnchors { + require.NotEmpty(t, deriveOpenAIAnchoredContentSessionSeed(body)) + } +} + +func TestDeriveOpenAIStablePrefixSessionSeed_RequiresMeaningfulPrefix(t *testing.T) { + tests := [][]byte{ + nil, + []byte(`{}`), + []byte(`{"model":"grok","input":"Question A"}`), + []byte(`{"model":"grok","tools":[],"input":"Question A"}`), + []byte(`{"model":"grok","functions":[],"instructions":" ","messages":[{"role":"system","content":""},{"role":"user","content":"Question A"}]}`), + } + + for _, body := range tests { + require.Empty(t, deriveOpenAIStablePrefixSessionSeed(body)) + } +} diff --git a/backend/internal/service/openai_gateway_grok_cache.go b/backend/internal/service/openai_gateway_grok_cache.go index 20934b94c3..1d689bce8a 100644 --- a/backend/internal/service/openai_gateway_grok_cache.go +++ b/backend/internal/service/openai_gateway_grok_cache.go @@ -42,7 +42,13 @@ func resolveGrokCacheIdentity(c *gin.Context, body []byte, explicitKey, upstream seed := explicitGrokCacheSeed(c, body, explicitKey) if seed == "" { - seed = deriveOpenAIContentSessionSeed(body) + seed = deriveOpenAIStablePrefixSessionSeed(body) + if seed == "" { + // A model alone is too broad for cache routing. Preserve the + // existing first-user-derived identity when no reusable prefix is + // available so unrelated prompts do not share one tenant-wide key. + seed = deriveOpenAIAnchoredContentSessionSeed(body) + } } if seed == "" { return "" diff --git a/backend/internal/service/openai_gateway_grok_cache_test.go b/backend/internal/service/openai_gateway_grok_cache_test.go index 556f19304f..42abfc5800 100644 --- a/backend/internal/service/openai_gateway_grok_cache_test.go +++ b/backend/internal/service/openai_gateway_grok_cache_test.go @@ -37,6 +37,63 @@ func TestResolveGrokCacheIdentityStableAcrossAppendOnlyTurns(t *testing.T) { require.Equal(t, first, second) } +func TestResolveGrokCacheIdentityStableAcrossIndependentPromptsWithSamePrefix(t *testing.T) { + gin.SetMode(gin.TestMode) + c := newGrokCacheTestContext(102) + firstBody := []byte(`{"model":"grok","instructions":"be concise","tools":[{"type":"function","name":"lookup"}],"input":[{"role":"user","content":"Question A"}]}`) + secondBody := []byte(`{"model":"grok","instructions":"be concise","tools":[{"type":"function","name":"lookup"}],"input":[{"role":"user","content":"Question B"}]}`) + + first := resolveGrokCacheIdentity(c, firstBody, "", "grok-4.5") + second := resolveGrokCacheIdentity(c, secondBody, "", "grok-4.5") + + require.NotEmpty(t, first) + require.Equal(t, first, second) +} + +func TestResolveGrokCacheIdentityStablePrefixIsolation(t *testing.T) { + gin.SetMode(gin.TestMode) + baseBody := []byte(`{"model":"grok","instructions":"be concise","tools":[{"type":"function","name":"lookup"}],"input":[{"role":"system","content":"System A"},{"role":"user","content":"Question A"}]}`) + differentInstructions := []byte(`{"model":"grok","instructions":"be detailed","tools":[{"type":"function","name":"lookup"}],"input":[{"role":"system","content":"System A"},{"role":"user","content":"Question B"}]}`) + differentSystem := []byte(`{"model":"grok","instructions":"be concise","tools":[{"type":"function","name":"lookup"}],"input":[{"role":"system","content":"System B"},{"role":"user","content":"Question B"}]}`) + differentTools := []byte(`{"model":"grok","instructions":"be concise","tools":[{"type":"function","name":"search"}],"input":[{"role":"system","content":"System A"},{"role":"user","content":"Question B"}]}`) + + base := resolveGrokCacheIdentity(newGrokCacheTestContext(103), baseBody, "", "grok-4.5") + require.NotEqual(t, base, resolveGrokCacheIdentity(newGrokCacheTestContext(104), baseBody, "", "grok-4.5")) + require.NotEqual(t, base, resolveGrokCacheIdentity(newGrokCacheTestContext(103), baseBody, "", "grok-4.3")) + require.NotEqual(t, base, resolveGrokCacheIdentity(newGrokCacheTestContext(103), differentInstructions, "", "grok-4.5")) + require.NotEqual(t, base, resolveGrokCacheIdentity(newGrokCacheTestContext(103), differentSystem, "", "grok-4.5")) + require.NotEqual(t, base, resolveGrokCacheIdentity(newGrokCacheTestContext(103), differentTools, "", "grok-4.5")) +} + +func TestResolveGrokCacheIdentityFallsBackWhenStablePrefixIsEmpty(t *testing.T) { + gin.SetMode(gin.TestMode) + c := newGrokCacheTestContext(105) + firstBody := []byte(`{"model":"grok","tools":[],"input":"Question A"}`) + secondBody := []byte(`{"model":"grok","tools":[],"input":"Question B"}`) + + first := resolveGrokCacheIdentity(c, firstBody, "", "grok-4.5") + second := resolveGrokCacheIdentity(c, secondBody, "", "grok-4.5") + + require.NotEmpty(t, first) + require.NotEmpty(t, second) + require.NotEqual(t, first, second) +} + +func TestResolveGrokCacheIdentitySkipsUnanchoredFallback(t *testing.T) { + gin.SetMode(gin.TestMode) + c := newGrokCacheTestContext(106) + tests := [][]byte{ + []byte(`{"model":"grok"}`), + []byte(`{"model":"grok","messages":[{"role":"assistant","content":"answer"}]}`), + []byte(`{"model":"grok","messages":[{"role":"user","content":""}]}`), + []byte(`{"model":"grok","input":" "}`), + } + + for _, body := range tests { + require.Empty(t, resolveGrokCacheIdentity(c, body, "", "grok-4.5")) + } +} + func TestResolveGrokCacheIdentityIsolatesAPIKeyAndMappedModel(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"grok","input":"same prompt"}`) diff --git a/backend/migrations/176_channel_monitor_grok_provider.sql b/backend/migrations/176_channel_monitor_grok_provider.sql new file mode 100644 index 0000000000..b1bad754a4 --- /dev/null +++ b/backend/migrations/176_channel_monitor_grok_provider.sql @@ -0,0 +1,39 @@ +-- Migration: 176_channel_monitor_grok_provider +-- Allow Grok as a channel-monitor provider. Grok checks use the existing +-- OpenAI-compatible chat completions protocol with model grok-4.5 by default. + +DO $$ +DECLARE + monitor_constraint_def TEXT; + template_constraint_def TEXT; +BEGIN + SELECT pg_get_constraintdef(c.oid) + INTO monitor_constraint_def + FROM pg_constraint c + JOIN pg_class t ON t.oid = c.conrelid + WHERE t.relname = 'channel_monitors' + AND c.conname = 'channel_monitors_provider_check'; + + IF monitor_constraint_def IS NULL OR position('grok' IN monitor_constraint_def) = 0 THEN + ALTER TABLE channel_monitors + DROP CONSTRAINT IF EXISTS channel_monitors_provider_check; + ALTER TABLE channel_monitors + ADD CONSTRAINT channel_monitors_provider_check + CHECK (provider IN ('openai', 'anthropic', 'gemini', 'grok')); + END IF; + + SELECT pg_get_constraintdef(c.oid) + INTO template_constraint_def + FROM pg_constraint c + JOIN pg_class t ON t.oid = c.conrelid + WHERE t.relname = 'channel_monitor_request_templates' + AND c.conname = 'channel_monitor_request_templates_provider_check'; + + IF template_constraint_def IS NULL OR position('grok' IN template_constraint_def) = 0 THEN + ALTER TABLE channel_monitor_request_templates + DROP CONSTRAINT IF EXISTS channel_monitor_request_templates_provider_check; + ALTER TABLE channel_monitor_request_templates + ADD CONSTRAINT channel_monitor_request_templates_provider_check + CHECK (provider IN ('openai', 'anthropic', 'gemini', 'grok')); + END IF; +END $$; diff --git a/backend/migrations/channel_monitor_grok_provider_migration_test.go b/backend/migrations/channel_monitor_grok_provider_migration_test.go new file mode 100644 index 0000000000..2545173f0e --- /dev/null +++ b/backend/migrations/channel_monitor_grok_provider_migration_test.go @@ -0,0 +1,20 @@ +package migrations + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestChannelMonitorGrokProviderMigration(t *testing.T) { + content, err := FS.ReadFile("176_channel_monitor_grok_provider.sql") + require.NoError(t, err) + + sql := strings.Join(strings.Fields(string(content)), " ") + require.Contains(t, sql, "channel_monitors_provider_check") + require.Contains(t, sql, "channel_monitor_request_templates_provider_check") + require.Contains(t, sql, "CHECK (provider IN ('openai', 'anthropic', 'gemini', 'grok'))") + require.Contains(t, sql, "position('grok' IN monitor_constraint_def) = 0") + require.Contains(t, sql, "position('grok' IN template_constraint_def) = 0") +} diff --git a/frontend/src/api/admin/channelMonitor.ts b/frontend/src/api/admin/channelMonitor.ts index 0b9c62231c..de605351e3 100644 --- a/frontend/src/api/admin/channelMonitor.ts +++ b/frontend/src/api/admin/channelMonitor.ts @@ -5,7 +5,7 @@ import { apiClient } from '../client' -export type Provider = 'openai' | 'anthropic' | 'gemini' +export type Provider = 'openai' | 'anthropic' | 'gemini' | 'grok' export type MonitorStatus = 'operational' | 'degraded' | 'failed' | 'error' export type BodyOverrideMode = 'off' | 'merge' | 'replace' export type APIMode = 'chat_completions' | 'responses' diff --git a/frontend/src/components/admin/monitor/MonitorAdvancedRequestConfig.vue b/frontend/src/components/admin/monitor/MonitorAdvancedRequestConfig.vue index 404b691692..c4ccdfa3ac 100644 --- a/frontend/src/components/admin/monitor/MonitorAdvancedRequestConfig.vue +++ b/frontend/src/components/admin/monitor/MonitorAdvancedRequestConfig.vue @@ -109,6 +109,8 @@ import { useI18n } from 'vue-i18n' import type { APIMode, BodyOverrideMode, Provider } from '@/api/admin/channelMonitor' import { API_MODE_RESPONSES, + DEFAULT_GROK_MODEL, + PROVIDER_GROK, PROVIDER_OPENAI, } from '@/constants/channelMonitor' @@ -305,11 +307,12 @@ const bodyPlaceholder = computed(() => { } return '{\n "model": "gpt-4o-mini",\n "instructions": "You are a health check endpoint. Reply briefly.",\n "input": "Reply with exactly: ok",\n "max_output_tokens": 20,\n "stream": false\n}' } - if (props.provider === PROVIDER_OPENAI) { + if (props.provider === PROVIDER_OPENAI || props.provider === PROVIDER_GROK) { if (props.bodyOverrideMode === 'merge') { return '{\n "max_tokens": 20\n}' } - return '{\n "model": "gpt-4o-mini",\n "messages": [{"role":"user","content":"Reply with exactly: ok"}],\n "max_tokens": 20,\n "stream": false\n}' + const model = props.provider === PROVIDER_GROK ? DEFAULT_GROK_MODEL : 'gpt-4o-mini' + return `{\n "model": "${model}",\n "messages": [{"role":"user","content":"Reply with exactly: ok"}],\n "max_tokens": 20,\n "stream": false\n}` } if (props.bodyOverrideMode === 'merge') { return '{\n "system": "You are Claude Code..."\n}' diff --git a/frontend/src/components/admin/monitor/MonitorFiltersBar.vue b/frontend/src/components/admin/monitor/MonitorFiltersBar.vue index eb2a5c7857..544238f49c 100644 --- a/frontend/src/components/admin/monitor/MonitorFiltersBar.vue +++ b/frontend/src/components/admin/monitor/MonitorFiltersBar.vue @@ -70,6 +70,7 @@ import { PROVIDER_OPENAI, PROVIDER_ANTHROPIC, PROVIDER_GEMINI, + PROVIDER_GROK, } from '@/constants/channelMonitor' defineProps<{ @@ -94,6 +95,7 @@ const providerFilterOptions = computed(() => [ { value: PROVIDER_OPENAI, label: t('monitorCommon.providers.openai') }, { value: PROVIDER_ANTHROPIC, label: t('monitorCommon.providers.anthropic') }, { value: PROVIDER_GEMINI, label: t('monitorCommon.providers.gemini') }, + { value: PROVIDER_GROK, label: t('monitorCommon.providers.grok') }, ]) const enabledFilterOptions = computed(() => [ diff --git a/frontend/src/components/admin/monitor/MonitorFormDialog.vue b/frontend/src/components/admin/monitor/MonitorFormDialog.vue index e6cab8edf1..14a9e2dd15 100644 --- a/frontend/src/components/admin/monitor/MonitorFormDialog.vue +++ b/frontend/src/components/admin/monitor/MonitorFormDialog.vue @@ -13,15 +13,16 @@
-
+
@@ -80,6 +81,7 @@ (() => [ { value: PROVIDER_ANTHROPIC, label: t('monitorCommon.providers.anthropic') }, { value: PROVIDER_OPENAI, label: t('monitorCommon.providers.openai') }, { value: PROVIDER_GEMINI, label: t('monitorCommon.providers.gemini') }, + { value: PROVIDER_GROK, label: t('monitorCommon.providers.grok') }, ]) +function selectProvider(provider: Provider) { + if (form.provider === provider) return + const previousProvider = form.provider + const clearGrokEndpoint = + previousProvider === PROVIDER_GROK && form.endpoint === DEFAULT_GROK_ENDPOINT + const clearGrokModel = + previousProvider === PROVIDER_GROK && form.primary_model === DEFAULT_GROK_MODEL + form.provider = provider + if (provider === PROVIDER_GROK) { + if (!form.endpoint.trim()) form.endpoint = DEFAULT_GROK_ENDPOINT + if (!form.primary_model.trim()) form.primary_model = DEFAULT_GROK_MODEL + return + } + if (clearGrokEndpoint) form.endpoint = '' + if (clearGrokModel) form.primary_model = '' +} + // Clear api_key whenever provider changes to avoid cross-provider key mismatch. // Editing mode loads api_key='' via loadFromMonitor and only sets it on user // typing, so clearing on provider change is always a safe no-op until the user diff --git a/frontend/src/components/admin/monitor/MonitorTemplateManagerDialog.vue b/frontend/src/components/admin/monitor/MonitorTemplateManagerDialog.vue index e54ecf631d..63b87db75a 100644 --- a/frontend/src/components/admin/monitor/MonitorTemplateManagerDialog.vue +++ b/frontend/src/components/admin/monitor/MonitorTemplateManagerDialog.vue @@ -7,7 +7,7 @@ >
-
+