From 72ccd1b11145a1d6b019716a691755b9cc93b32c Mon Sep 17 00:00:00 2001 From: jianjian Date: Thu, 2 Jul 2026 20:22:17 +0000 Subject: [PATCH 01/28] fix: batch group capacity summaries --- backend/internal/repository/account_repo.go | 89 +++++++++ backend/internal/repository/group_repo.go | 43 +++++ .../service/group_capacity_service.go | 180 +++++++++++++++++- .../service/group_capacity_service_test.go | 179 +++++++++++++++++ 4 files changed, 487 insertions(+), 4 deletions(-) create mode 100644 backend/internal/service/group_capacity_service_test.go diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 2a2f5e0576..be03a6225b 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -1026,6 +1026,95 @@ func (r *accountRepository) ListSchedulableByGroupID(ctx context.Context, groupI }) } +func (r *accountRepository) ListSchedulableCapacityByGroupIDs(ctx context.Context, groupIDs []int64) ([]service.GroupAccountCapacityRow, error) { + groupIDs = uniquePositiveInt64s(groupIDs) + if len(groupIDs) == 0 { + return []service.GroupAccountCapacityRow{}, nil + } + if r.sql == nil { + rows := make([]service.GroupAccountCapacityRow, 0) + for _, groupID := range groupIDs { + accounts, err := r.ListSchedulableByGroupID(ctx, groupID) + if err != nil { + return nil, err + } + for i := range accounts { + acc := &accounts[i] + rows = append(rows, service.GroupAccountCapacityRow{ + GroupID: groupID, + AccountID: acc.ID, + Concurrency: acc.Concurrency, + Extra: copyJSONMap(acc.Extra), + SessionWindowStart: acc.SessionWindowStart, + SessionWindowEnd: acc.SessionWindowEnd, + SessionWindowStatus: acc.SessionWindowStatus, + }) + } + } + return rows, nil + } + + rows, err := r.sql.QueryContext(ctx, ` + SELECT + ag.group_id, + a.id AS account_id, + a.concurrency, + COALESCE(a.extra, '{}'::jsonb)::text AS extra, + a.session_window_start, + a.session_window_end, + COALESCE(a.session_window_status, '') AS session_window_status + FROM account_groups ag + JOIN accounts a ON a.id = ag.account_id + LEFT JOIN proxies p + ON p.id = a.proxy_id + AND p.deleted_at IS NULL + AND p.status = $2 + WHERE ag.group_id = ANY($1) + AND a.deleted_at IS NULL + AND a.status = $2 + AND a.schedulable = TRUE + AND (a.owner_user_id IS NULL OR p.id IS NOT NULL) + AND (a.temp_unschedulable_until IS NULL OR a.temp_unschedulable_until <= $3) + AND (a.expires_at IS NULL OR a.expires_at > $3 OR a.auto_pause_on_expired = FALSE) + AND (a.overload_until IS NULL OR a.overload_until <= $3) + AND (a.rate_limit_reset_at IS NULL OR a.rate_limit_reset_at <= $3) + ORDER BY ag.group_id ASC, ag.priority ASC, a.priority ASC, a.id ASC + `, pq.Array(groupIDs), service.StatusActive, time.Now()) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + out := make([]service.GroupAccountCapacityRow, 0) + for rows.Next() { + var row service.GroupAccountCapacityRow + var extraRaw string + if err := rows.Scan( + &row.GroupID, + &row.AccountID, + &row.Concurrency, + &extraRaw, + &row.SessionWindowStart, + &row.SessionWindowEnd, + &row.SessionWindowStatus, + ); err != nil { + return nil, err + } + if extraRaw != "" && extraRaw != "null" { + var extra map[string]any + if err := json.Unmarshal([]byte(extraRaw), &extra); err != nil { + return nil, err + } + row.Extra = extra + } + out = append(out, row) + } + if err := rows.Err(); err != nil { + return nil, err + } + return out, nil +} + func (r *accountRepository) ListSchedulableByPlatform(ctx context.Context, platform string) ([]service.Account, error) { now := time.Now() accounts, err := r.client.Account.Query(). diff --git a/backend/internal/repository/group_repo.go b/backend/internal/repository/group_repo.go index 4e839b6a12..a4e173006e 100644 --- a/backend/internal/repository/group_repo.go +++ b/backend/internal/repository/group_repo.go @@ -466,6 +466,49 @@ func (r *groupRepository) ListActive(ctx context.Context) ([]service.Group, erro return outGroups, nil } +func (r *groupRepository) ListActiveIDs(ctx context.Context) ([]int64, error) { + if r.sql != nil { + rows, err := r.sql.QueryContext(ctx, ` + SELECT id + FROM groups + WHERE status = $1 + AND deleted_at IS NULL + ORDER BY sort_order ASC, id ASC + `, service.StatusActive) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + ids := make([]int64, 0) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return ids, nil + } + + groups, err := r.client.Group.Query(). + Where(group.StatusEQ(service.StatusActive)). + Select(group.FieldID). + Order(dbent.Asc(group.FieldSortOrder), dbent.Asc(group.FieldID)). + All(ctx) + if err != nil { + return nil, err + } + ids := make([]int64, 0, len(groups)) + for i := range groups { + ids = append(ids, groups[i].ID) + } + return ids, nil +} + func (r *groupRepository) ListActiveByPlatform(ctx context.Context, platform string) ([]service.Group, error) { groups, err := r.client.Group.Query(). Where(group.StatusEQ(service.StatusActive), group.PlatformEQ(platform)). diff --git a/backend/internal/service/group_capacity_service.go b/backend/internal/service/group_capacity_service.go index 459084dc59..b77b35773b 100644 --- a/backend/internal/service/group_capacity_service.go +++ b/backend/internal/service/group_capacity_service.go @@ -16,6 +16,26 @@ type GroupCapacitySummary struct { RPMMax int `json:"rpm_max"` } +// GroupAccountCapacityRow is the lightweight account projection needed for +// capacity summary aggregation. +type GroupAccountCapacityRow struct { + GroupID int64 + AccountID int64 + Concurrency int + Extra map[string]any + SessionWindowStart *time.Time + SessionWindowEnd *time.Time + SessionWindowStatus string +} + +type groupCapacityActiveGroupIDLister interface { + ListActiveIDs(ctx context.Context) ([]int64, error) +} + +type groupCapacityAccountLister interface { + ListSchedulableCapacityByGroupIDs(ctx context.Context, groupIDs []int64) ([]GroupAccountCapacityRow, error) +} + // GroupCapacityService aggregates per-group capacity from runtime data. type GroupCapacityService struct { accountRepo AccountRepository @@ -44,24 +64,176 @@ func NewGroupCapacityService( // GetAllGroupCapacity returns capacity summary for all active groups. func (s *GroupCapacityService) GetAllGroupCapacity(ctx context.Context) ([]GroupCapacitySummary, error) { - groups, err := s.groupRepo.ListActive(ctx) + groupIDs, err := s.listActiveGroupIDs(ctx) if err != nil { return nil, err } - results := make([]GroupCapacitySummary, 0, len(groups)) + if lister, ok := s.accountRepo.(groupCapacityAccountLister); ok { + return s.getGroupCapacitiesBatch(ctx, groupIDs, lister) + } + + return s.getGroupCapacitiesSequential(ctx, groupIDs), nil +} + +func (s *GroupCapacityService) listActiveGroupIDs(ctx context.Context) ([]int64, error) { + if lister, ok := s.groupRepo.(groupCapacityActiveGroupIDLister); ok { + return lister.ListActiveIDs(ctx) + } + + groups, err := s.groupRepo.ListActive(ctx) + if err != nil { + return nil, err + } + groupIDs := make([]int64, 0, len(groups)) for i := range groups { - cap, err := s.getGroupCapacity(ctx, groups[i].ID) + groupIDs = append(groupIDs, groups[i].ID) + } + return groupIDs, nil +} + +func (s *GroupCapacityService) getGroupCapacitiesSequential(ctx context.Context, groupIDs []int64) []GroupCapacitySummary { + results := make([]GroupCapacitySummary, 0, len(groupIDs)) + for _, groupID := range groupIDs { + cap, err := s.getGroupCapacity(ctx, groupID) if err != nil { // Skip groups with errors, return partial results continue } - cap.GroupID = groups[i].ID + cap.GroupID = groupID results = append(results, cap) } + return results +} + +type groupCapacityAccountRef struct { + groupID int64 + accountID int64 +} + +func (s *GroupCapacityService) getGroupCapacitiesBatch(ctx context.Context, groupIDs []int64, lister groupCapacityAccountLister) ([]GroupCapacitySummary, error) { + results := make([]GroupCapacitySummary, len(groupIDs)) + groupIndex := make(map[int64]int, len(groupIDs)) + for i, groupID := range groupIDs { + results[i].GroupID = groupID + groupIndex[groupID] = i + } + if len(groupIDs) == 0 { + return results, nil + } + + rows, err := lister.ListSchedulableCapacityByGroupIDs(ctx, groupIDs) + if err != nil { + return nil, err + } + if len(rows) == 0 { + return results, nil + } + + refs := make([]groupCapacityAccountRef, 0, len(rows)) + seenGroupAccount := make(map[groupCapacityAccountRef]struct{}, len(rows)) + accountIDSet := make(map[int64]struct{}, len(rows)) + accountIDs := make([]int64, 0, len(rows)) + sessionTimeouts := make(map[int64]time.Duration) + + for _, row := range rows { + idx, ok := groupIndex[row.GroupID] + if !ok || row.AccountID <= 0 { + continue + } + + ref := groupCapacityAccountRef{groupID: row.GroupID, accountID: row.AccountID} + if _, ok := seenGroupAccount[ref]; ok { + continue + } + seenGroupAccount[ref] = struct{}{} + refs = append(refs, ref) + + if _, ok := accountIDSet[row.AccountID]; !ok { + accountIDSet[row.AccountID] = struct{}{} + accountIDs = append(accountIDs, row.AccountID) + } + + acc := Account{ + ID: row.AccountID, + Concurrency: row.Concurrency, + Extra: row.Extra, + SessionWindowStart: row.SessionWindowStart, + SessionWindowEnd: row.SessionWindowEnd, + SessionWindowStatus: row.SessionWindowStatus, + } + + results[idx].ConcurrencyMax += acc.Concurrency + + if maxSessions := acc.GetMaxSessions(); maxSessions > 0 { + results[idx].SessionsMax += maxSessions + timeout := time.Duration(acc.GetSessionIdleTimeoutMinutes()) * time.Minute + if timeout <= 0 { + timeout = 5 * time.Minute + } + sessionTimeouts[acc.ID] = timeout + } + + if rpm := acc.GetBaseRPM(); rpm > 0 { + results[idx].RPMMax += rpm + } + } + + if len(accountIDs) == 0 { + return results, nil + } + + concurrencyMap := map[int64]int{} + if s.concurrencyService != nil { + concurrencyMap, _ = s.concurrencyService.GetAccountConcurrencyBatch(ctx, accountIDs) + } + + sessionAccountIDs := accountIDsForGroupsWithLimit(refs, groupIndex, results, func(summary GroupCapacitySummary) bool { + return summary.SessionsMax > 0 + }) + var sessionsMap map[int64]int + if len(sessionAccountIDs) > 0 && s.sessionLimitCache != nil { + sessionsMap, _ = s.sessionLimitCache.GetActiveSessionCountBatch(ctx, sessionAccountIDs, sessionTimeouts) + } + + rpmAccountIDs := accountIDsForGroupsWithLimit(refs, groupIndex, results, func(summary GroupCapacitySummary) bool { + return summary.RPMMax > 0 + }) + var rpmMap map[int64]int + if len(rpmAccountIDs) > 0 && s.rpmCache != nil { + rpmMap, _ = s.rpmCache.GetRPMBatch(ctx, rpmAccountIDs) + } + + for _, ref := range refs { + idx := groupIndex[ref.groupID] + results[idx].ConcurrencyUsed += concurrencyMap[ref.accountID] + if sessionsMap != nil && results[idx].SessionsMax > 0 { + results[idx].SessionsUsed += sessionsMap[ref.accountID] + } + if rpmMap != nil && results[idx].RPMMax > 0 { + results[idx].RPMUsed += rpmMap[ref.accountID] + } + } return results, nil } +func accountIDsForGroupsWithLimit(refs []groupCapacityAccountRef, groupIndex map[int64]int, summaries []GroupCapacitySummary, include func(GroupCapacitySummary) bool) []int64 { + seen := make(map[int64]struct{}) + accountIDs := make([]int64, 0) + for _, ref := range refs { + idx, ok := groupIndex[ref.groupID] + if !ok || !include(summaries[idx]) { + continue + } + if _, ok := seen[ref.accountID]; ok { + continue + } + seen[ref.accountID] = struct{}{} + accountIDs = append(accountIDs, ref.accountID) + } + return accountIDs +} + func (s *GroupCapacityService) getGroupCapacity(ctx context.Context, groupID int64) (GroupCapacitySummary, error) { accounts, err := s.accountRepo.ListSchedulableByGroupID(ctx, groupID) if err != nil { diff --git a/backend/internal/service/group_capacity_service_test.go b/backend/internal/service/group_capacity_service_test.go new file mode 100644 index 0000000000..73927307d2 --- /dev/null +++ b/backend/internal/service/group_capacity_service_test.go @@ -0,0 +1,179 @@ +package service + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type groupCapacityAccountRepoStub struct { + AccountRepository + rows []GroupAccountCapacityRow + requested []int64 +} + +func (s *groupCapacityAccountRepoStub) ListSchedulableCapacityByGroupIDs(_ context.Context, groupIDs []int64) ([]GroupAccountCapacityRow, error) { + s.requested = append([]int64(nil), groupIDs...) + return append([]GroupAccountCapacityRow(nil), s.rows...), nil +} + +type groupCapacityGroupRepoStub struct { + GroupRepository + groupIDs []int64 + listCalls int +} + +func (s *groupCapacityGroupRepoStub) ListActiveIDs(context.Context) ([]int64, error) { + s.listCalls++ + return append([]int64(nil), s.groupIDs...), nil +} + +type groupCapacityConcurrencyCacheStub struct { + ConcurrencyCache + counts map[int64]int + requested []int64 +} + +func (s *groupCapacityConcurrencyCacheStub) GetAccountConcurrencyBatch(_ context.Context, accountIDs []int64) (map[int64]int, error) { + s.requested = append([]int64(nil), accountIDs...) + out := make(map[int64]int, len(accountIDs)) + for _, id := range accountIDs { + out[id] = s.counts[id] + } + return out, nil +} + +type groupCapacitySessionCacheStub struct { + SessionLimitCache + counts map[int64]int + requested []int64 + idleTimeouts map[int64]time.Duration +} + +func (s *groupCapacitySessionCacheStub) GetActiveSessionCountBatch(_ context.Context, accountIDs []int64, idleTimeouts map[int64]time.Duration) (map[int64]int, error) { + s.requested = append([]int64(nil), accountIDs...) + s.idleTimeouts = make(map[int64]time.Duration, len(idleTimeouts)) + for id, timeout := range idleTimeouts { + s.idleTimeouts[id] = timeout + } + out := make(map[int64]int, len(accountIDs)) + for _, id := range accountIDs { + out[id] = s.counts[id] + } + return out, nil +} + +type groupCapacityRPMCacheStub struct { + RPMCache + counts map[int64]int + requested []int64 +} + +func (s *groupCapacityRPMCacheStub) GetRPMBatch(_ context.Context, accountIDs []int64) (map[int64]int, error) { + s.requested = append([]int64(nil), accountIDs...) + out := make(map[int64]int, len(accountIDs)) + for _, id := range accountIDs { + out[id] = s.counts[id] + } + return out, nil +} + +func TestGetAllGroupCapacityBatchAggregatesRuntimeAndLimits(t *testing.T) { + accountRepo := &groupCapacityAccountRepoStub{ + rows: []GroupAccountCapacityRow{ + { + GroupID: 10, + AccountID: 1, + Concurrency: 2, + Extra: map[string]any{ + "max_sessions": 3, + "session_idle_timeout_minutes": 7, + "base_rpm": 11, + }, + }, + { + GroupID: 20, + AccountID: 1, + Concurrency: 2, + Extra: map[string]any{ + "max_sessions": 3, + "session_idle_timeout_minutes": 7, + "base_rpm": 11, + }, + }, + { + GroupID: 20, + AccountID: 2, + Concurrency: 4, + Extra: map[string]any{ + "max_sessions": 1, + "session_idle_timeout_minutes": 9, + "base_rpm": 13, + }, + }, + }, + } + groupRepo := &groupCapacityGroupRepoStub{groupIDs: []int64{10, 20}} + concurrencyCache := &groupCapacityConcurrencyCacheStub{counts: map[int64]int{1: 1, 2: 2}} + sessionCache := &groupCapacitySessionCacheStub{counts: map[int64]int{1: 2, 2: 1}} + rpmCache := &groupCapacityRPMCacheStub{counts: map[int64]int{1: 5, 2: 7}} + svc := NewGroupCapacityService( + accountRepo, + groupRepo, + NewConcurrencyService(concurrencyCache), + sessionCache, + rpmCache, + ) + + results, err := svc.GetAllGroupCapacity(context.Background()) + require.NoError(t, err) + + require.Equal(t, 1, groupRepo.listCalls) + require.Equal(t, []int64{10, 20}, accountRepo.requested) + require.Equal(t, []int64{1, 2}, concurrencyCache.requested) + require.ElementsMatch(t, []int64{1, 2}, sessionCache.requested) + require.ElementsMatch(t, []int64{1, 2}, rpmCache.requested) + require.Equal(t, 7*time.Minute, sessionCache.idleTimeouts[1]) + require.Equal(t, 9*time.Minute, sessionCache.idleTimeouts[2]) + + require.Equal(t, []GroupCapacitySummary{ + { + GroupID: 10, + ConcurrencyUsed: 1, + ConcurrencyMax: 2, + SessionsUsed: 2, + SessionsMax: 3, + RPMUsed: 5, + RPMMax: 11, + }, + { + GroupID: 20, + ConcurrencyUsed: 3, + ConcurrencyMax: 6, + SessionsUsed: 3, + SessionsMax: 4, + RPMUsed: 12, + RPMMax: 24, + }, + }, results) +} + +func TestGetAllGroupCapacityBatchKeepsEmptyGroupRows(t *testing.T) { + accountRepo := &groupCapacityAccountRepoStub{ + rows: []GroupAccountCapacityRow{ + {GroupID: 20, AccountID: 2, Concurrency: 4}, + }, + } + groupRepo := &groupCapacityGroupRepoStub{groupIDs: []int64{10, 20}} + svc := NewGroupCapacityService(accountRepo, groupRepo, nil, nil, nil) + + results, err := svc.GetAllGroupCapacity(context.Background()) + require.NoError(t, err) + + require.Equal(t, []GroupCapacitySummary{ + {GroupID: 10}, + {GroupID: 20, ConcurrencyMax: 4}, + }, results) +} From 36d5f4e4caf87fe88c7c52445c22c3a83d42b56a Mon Sep 17 00:00:00 2001 From: weiness <840833211@qq.com> Date: Fri, 3 Jul 2026 10:29:33 +0800 Subject: [PATCH 02/28] feat: make setup migration timeout configurable --- backend/internal/setup/setup.go | 26 ++++++++++++++++++-------- backend/internal/setup/setup_test.go | 17 +++++++++++++++++ deploy/.env.example | 7 +++++++ deploy/docker-compose.dev.yml | 1 + deploy/docker-compose.local.yml | 5 +++++ deploy/docker-compose.standalone.yml | 5 +++++ deploy/docker-compose.yml | 5 +++++ 7 files changed, 58 insertions(+), 8 deletions(-) diff --git a/backend/internal/setup/setup.go b/backend/internal/setup/setup.go index 51baf3dfe6..a2c4e2847a 100644 --- a/backend/internal/setup/setup.go +++ b/backend/internal/setup/setup.go @@ -28,6 +28,7 @@ const ( InstallLockFile = ".installed" defaultUserConcurrency = 5 simpleModeAdminConcurrency = 30 + defaultMigrationTimeout = 60 * time.Second ) func setupDefaultAdminConcurrency() int { @@ -73,12 +74,13 @@ func GetInstallLockPath() string { // SetupConfig holds the setup configuration type SetupConfig struct { - Database DatabaseConfig `json:"database" yaml:"database"` - Redis RedisConfig `json:"redis" yaml:"redis"` - Admin AdminConfig `json:"admin" yaml:"-"` // Not stored in config file - Server ServerConfig `json:"server" yaml:"server"` - JWT JWTConfig `json:"jwt" yaml:"jwt"` - Timezone string `json:"timezone" yaml:"timezone"` // e.g. "Asia/Shanghai", "UTC" + Database DatabaseConfig `json:"database" yaml:"database"` + Redis RedisConfig `json:"redis" yaml:"redis"` + Admin AdminConfig `json:"admin" yaml:"-"` // Not stored in config file + Server ServerConfig `json:"server" yaml:"server"` + JWT JWTConfig `json:"jwt" yaml:"jwt"` + Timezone string `json:"timezone" yaml:"timezone"` // e.g. "Asia/Shanghai", "UTC" + MigrationTimeoutSeconds int `json:"migration_timeout_seconds" yaml:"migration_timeout_seconds,omitempty"` } type DatabaseConfig struct { @@ -350,11 +352,18 @@ func initializeDatabase(cfg *SetupConfig) error { } }() - migrationCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + migrationCtx, cancel := context.WithTimeout(context.Background(), cfg.migrationTimeout()) defer cancel() return repository.ApplyMigrations(migrationCtx, db) } +func (cfg *SetupConfig) migrationTimeout() time.Duration { + if cfg != nil && cfg.MigrationTimeoutSeconds > 0 { + return time.Duration(cfg.MigrationTimeoutSeconds) * time.Second + } + return defaultMigrationTimeout +} + func createAdminUser(cfg *SetupConfig) (bool, string, error) { dsn := fmt.Sprintf( "host=%s port=%d user=%s password=%s dbname=%s sslmode=%s", @@ -578,7 +587,8 @@ func AutoSetupFromEnv() error { Secret: getEnvOrDefault("JWT_SECRET", ""), ExpireHour: getEnvIntOrDefault("JWT_EXPIRE_HOUR", 24), }, - Timezone: tz, + Timezone: tz, + MigrationTimeoutSeconds: getEnvIntOrDefault("SETUP_MIGRATION_TIMEOUT_SECONDS", 0), } // Generate JWT secret if not provided diff --git a/backend/internal/setup/setup_test.go b/backend/internal/setup/setup_test.go index a2aa2f4cc1..b95c162bda 100644 --- a/backend/internal/setup/setup_test.go +++ b/backend/internal/setup/setup_test.go @@ -4,6 +4,7 @@ import ( "os" "strings" "testing" + "time" ) func TestDecideAdminBootstrap(t *testing.T) { @@ -70,6 +71,22 @@ func TestSetupDefaultAdminConcurrency(t *testing.T) { }) } +func TestSetupMigrationTimeout(t *testing.T) { + t.Run("uses default timeout when unset", func(t *testing.T) { + cfg := &SetupConfig{} + if got := cfg.migrationTimeout(); got != 60*time.Second { + t.Fatalf("migrationTimeout()=%s, want 60s", got) + } + }) + + t.Run("uses configured timeout", func(t *testing.T) { + cfg := &SetupConfig{MigrationTimeoutSeconds: 300} + if got := cfg.migrationTimeout(); got != 300*time.Second { + t.Fatalf("migrationTimeout()=%s, want 300s", got) + } + }) +} + func TestWriteConfigFileKeepsDefaultUserConcurrency(t *testing.T) { t.Setenv("RUN_MODE", "simple") t.Setenv("DATA_DIR", t.TempDir()) diff --git a/deploy/.env.example b/deploy/.env.example index 59e4b44b91..d8892dbecf 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -196,6 +196,13 @@ JWT_EXPIRE_HOUR=24 # - =0: 回退使用 JWT_EXPIRE_HOUR JWT_ACCESS_TOKEN_EXPIRE_MINUTES=0 +# ----------------------------------------------------------------------------- +# Setup Configuration +# ----------------------------------------------------------------------------- +# Database migration timeout during initial setup, in seconds. +# Leave 0 to use the built-in default of 60 seconds. +SETUP_MIGRATION_TIMEOUT_SECONDS=0 + # ----------------------------------------------------------------------------- # TOTP (2FA) Configuration # TOTP(双因素认证)配置 diff --git a/deploy/docker-compose.dev.yml b/deploy/docker-compose.dev.yml index 7755fdbeff..07e89e0b51 100644 --- a/deploy/docker-compose.dev.yml +++ b/deploy/docker-compose.dev.yml @@ -38,6 +38,7 @@ services: - ADMIN_EMAIL=${ADMIN_EMAIL:-admin@sub2api.local} - ADMIN_PASSWORD=${ADMIN_PASSWORD:-} - JWT_SECRET=${JWT_SECRET:-} + - SETUP_MIGRATION_TIMEOUT_SECONDS=${SETUP_MIGRATION_TIMEOUT_SECONDS:-0} - TOTP_ENCRYPTION_KEY=${TOTP_ENCRYPTION_KEY:-} - TZ=${TZ:-Asia/Shanghai} # OpenAI HTTP upstream protocol/timeout diff --git a/deploy/docker-compose.local.yml b/deploy/docker-compose.local.yml index b15be2402d..21f46e3760 100644 --- a/deploy/docker-compose.local.yml +++ b/deploy/docker-compose.local.yml @@ -94,6 +94,11 @@ services: - JWT_SECRET=${JWT_SECRET:-} - JWT_EXPIRE_HOUR=${JWT_EXPIRE_HOUR:-24} + # ======================================================================= + # Setup Configuration + # ======================================================================= + - SETUP_MIGRATION_TIMEOUT_SECONDS=${SETUP_MIGRATION_TIMEOUT_SECONDS:-0} + # ======================================================================= # TOTP (2FA) Configuration # ======================================================================= diff --git a/deploy/docker-compose.standalone.yml b/deploy/docker-compose.standalone.yml index 32afb28d6c..2e1d335624 100644 --- a/deploy/docker-compose.standalone.yml +++ b/deploy/docker-compose.standalone.yml @@ -76,6 +76,11 @@ services: - JWT_SECRET=${JWT_SECRET:-} - JWT_EXPIRE_HOUR=${JWT_EXPIRE_HOUR:-24} + # ======================================================================= + # Setup Configuration + # ======================================================================= + - SETUP_MIGRATION_TIMEOUT_SECONDS=${SETUP_MIGRATION_TIMEOUT_SECONDS:-0} + # ======================================================================= # Timezone Configuration # ======================================================================= diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index fd682a87e6..3da3bcbfb0 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -90,6 +90,11 @@ services: - JWT_SECRET=${JWT_SECRET:-} - JWT_EXPIRE_HOUR=${JWT_EXPIRE_HOUR:-24} + # ======================================================================= + # Setup Configuration + # ======================================================================= + - SETUP_MIGRATION_TIMEOUT_SECONDS=${SETUP_MIGRATION_TIMEOUT_SECONDS:-0} + # ======================================================================= # TOTP (2FA) Configuration # ======================================================================= From 2b462b07debde257dff50ca59ee4d30b75b7be80 Mon Sep 17 00:00:00 2001 From: Heatherm Huang Date: Fri, 3 Jul 2026 14:41:18 +0800 Subject: [PATCH 03/28] fix: preserve configured Grok OAuth concurrency --- .../service/admin_account_concurrency_test.go | 17 +++++------------ backend/internal/service/admin_service.go | 3 --- 2 files changed, 5 insertions(+), 15 deletions(-) diff --git a/backend/internal/service/admin_account_concurrency_test.go b/backend/internal/service/admin_account_concurrency_test.go index 3544f80e24..da57b5a64f 100644 --- a/backend/internal/service/admin_account_concurrency_test.go +++ b/backend/internal/service/admin_account_concurrency_test.go @@ -5,23 +5,16 @@ package service import ( "testing" - "github.com/Wei-Shaw/sub2api/internal/pkg/xai" "github.com/stretchr/testify/require" ) -func TestNormalizeAccountConcurrencyCapsGrokOAuthUnlessUnsafe(t *testing.T) { - t.Setenv(xai.EnvUnsafeAllowHighConcurrency, "") - +func TestNormalizeAccountConcurrencyDefaultsInvalidGrokOAuthToOne(t *testing.T) { require.Equal(t, 1, normalizeAccountConcurrency(PlatformGrok, AccountTypeOAuth, 0)) require.Equal(t, 1, normalizeAccountConcurrency(PlatformGrok, AccountTypeOAuth, -5)) - require.Equal(t, 1, normalizeAccountConcurrency(PlatformGrok, AccountTypeOAuth, 50)) +} + +func TestNormalizeAccountConcurrencyPreservesExplicitValues(t *testing.T) { + require.Equal(t, 50, normalizeAccountConcurrency(PlatformGrok, AccountTypeOAuth, 50)) require.Equal(t, 2, normalizeAccountConcurrency(PlatformOpenAI, AccountTypeOAuth, 2)) require.Equal(t, 2, normalizeAccountConcurrency(PlatformGrok, AccountTypeAPIKey, 2)) } - -func TestNormalizeAccountConcurrencyAllowsGrokOAuthUnsafeOverride(t *testing.T) { - t.Setenv(xai.EnvUnsafeAllowHighConcurrency, "true") - - require.Equal(t, 50, normalizeAccountConcurrency(PlatformGrok, AccountTypeOAuth, 50)) - require.Equal(t, 1, normalizeAccountConcurrency(PlatformGrok, AccountTypeOAuth, 0)) -} diff --git a/backend/internal/service/admin_service.go b/backend/internal/service/admin_service.go index bacd134db4..7685174f3b 100644 --- a/backend/internal/service/admin_service.go +++ b/backend/internal/service/admin_service.go @@ -2640,9 +2640,6 @@ func normalizeAccountConcurrency(platform, accountType string, concurrency int) if concurrency <= 0 { return 1 } - if concurrency > 1 && !xai.AllowUnsafeHighConcurrency() { - return 1 - } } return concurrency } From c620ad6a36376d8c010950d17d7d6f373e8b67a0 Mon Sep 17 00:00:00 2001 From: jianjian Date: Fri, 3 Jul 2026 08:44:39 +0000 Subject: [PATCH 04/28] fix: align group capacity SQL with target schema --- backend/internal/repository/account_repo.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index be03a6225b..18f25a5949 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -1065,15 +1065,10 @@ func (r *accountRepository) ListSchedulableCapacityByGroupIDs(ctx context.Contex COALESCE(a.session_window_status, '') AS session_window_status FROM account_groups ag JOIN accounts a ON a.id = ag.account_id - LEFT JOIN proxies p - ON p.id = a.proxy_id - AND p.deleted_at IS NULL - AND p.status = $2 WHERE ag.group_id = ANY($1) AND a.deleted_at IS NULL AND a.status = $2 AND a.schedulable = TRUE - AND (a.owner_user_id IS NULL OR p.id IS NOT NULL) AND (a.temp_unschedulable_until IS NULL OR a.temp_unschedulable_until <= $3) AND (a.expires_at IS NULL OR a.expires_at > $3 OR a.auto_pause_on_expired = FALSE) AND (a.overload_until IS NULL OR a.overload_until <= $3) From b2687b22589c2f7aa12f4092bb89ea3739fc7735 Mon Sep 17 00:00:00 2001 From: Heatherm Huang Date: Fri, 3 Jul 2026 19:43:54 +0800 Subject: [PATCH 05/28] fix(frontend): edit grok oauth model mapping --- README.md | 6 +- .../components/account/EditAccountModal.vue | 23 +++++--- .../__tests__/EditAccountModal.spec.ts | 55 +++++++++++++++++++ frontend/src/composables/useModelWhitelist.ts | 13 ++++- 4 files changed, 86 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index d960d93fa1..21ea27dee3 100644 --- a/README.md +++ b/README.md @@ -630,8 +630,10 @@ Sub2API supports Grok subscription accounts through xAI OAuth and forwards OpenA - Public Claude-compatible target: `/v1/messages`, converted to xAI Responses and returned as Anthropic Messages output for Claude CLI style clients - Public Chat Completions targets: `/v1/chat/completions` and `/chat/completions`, forwarded to `${XAI_BASE_URL:-https://api.x.ai/v1}/chat/completions` - Codex CLI style Responses WebSocket ingress is accepted on the Responses targets and bridged to xAI HTTP/SSE Responses upstream -- Initial models: `grok-4.3`, `grok-build-0.1`, `grok-4.20-0309-reasoning`, `grok-4.20-0309-non-reasoning`, and `grok-4.20-multi-agent-0309` -- Out of scope for this provider: image, video, TTS, transcription, browser automation, cookies, and Grok web scraping +- Initial text models: `grok-4.3`, `grok-build-0.1`, `grok-4.20-0309-reasoning`, `grok-4.20-0309-non-reasoning`, and `grok-4.20-multi-agent-0309` +- Media targets for Grok groups: `/v1/images/generations`, `/images/generations`, `/v1/images/edits`, `/images/edits`, `/v1/videos/generations`, `/videos/generations`, `/v1/videos/{request_id}`, and `/videos/{request_id}`. Generation requests require the group image-generation permission. +- Media models: `grok-imagine`, `grok-imagine-image-quality`, `grok-imagine-image`, `grok-imagine-edit`, `grok-imagine-video`, and `grok-imagine-video-1.5` +- Out of scope for this provider: TTS, transcription, browser automation, cookies, and Grok web scraping ### OAuth Configuration diff --git a/frontend/src/components/account/EditAccountModal.vue b/frontend/src/components/account/EditAccountModal.vue index b5f97753d8..cf65d96709 100644 --- a/frontend/src/components/account/EditAccountModal.vue +++ b/frontend/src/components/account/EditAccountModal.vue @@ -419,9 +419,9 @@ - +
@@ -3244,8 +3244,8 @@ const syncFormFromAccount = (newAccount: Account | null) => { : 'https://api.anthropic.com' editBaseUrl.value = platformDefaultUrl - // Load model mappings for OpenAI OAuth accounts - if (newAccount.platform === 'openai' && newAccount.credentials) { + // Load model mappings for OpenAI/Grok OAuth accounts + if ((newAccount.platform === 'openai' || newAccount.platform === 'grok') && newAccount.credentials) { const oauthCredentials = newAccount.credentials as Record loadModelRestrictionFromMapping(oauthCredentials.model_mapping as Record | undefined) } else { @@ -3972,14 +3972,23 @@ const handleSubmit = async () => { updatePayload.credentials = newCredentials } - // OpenAI OAuth: persist model mapping to credentials - if (props.account.platform === 'openai' && props.account.type === 'oauth') { + // OpenAI/Grok OAuth: persist model mapping to credentials + if ((props.account.platform === 'openai' || props.account.platform === 'grok') && props.account.type === 'oauth') { const currentCredentials = isSparkShadow.value ? {} : (updatePayload.credentials as Record) || ((props.account.credentials as Record) || {}) const newCredentials: Record = { ...currentCredentials } - applyOpenAIModelMappingCredentials(newCredentials) + if (props.account.platform === 'openai') { + applyOpenAIModelMappingCredentials(newCredentials) + } else { + const modelMapping = buildModelRestrictionMapping() + if (modelMapping) { + newCredentials.model_mapping = modelMapping + } else { + delete newCredentials.model_mapping + } + } updatePayload.credentials = newCredentials } diff --git a/frontend/src/components/account/__tests__/EditAccountModal.spec.ts b/frontend/src/components/account/__tests__/EditAccountModal.spec.ts index 381d512ee7..5cc22a2958 100644 --- a/frontend/src/components/account/__tests__/EditAccountModal.spec.ts +++ b/frontend/src/components/account/__tests__/EditAccountModal.spec.ts @@ -241,6 +241,32 @@ function buildAntigravityAccount(projectId = 'configured-project') { } as any } +function buildGrokOAuthAccount() { + return { + id: 5, + name: 'Grok OAuth', + notes: '', + platform: 'grok', + type: 'oauth', + credentials: { + refresh_token: 'grok-rt', + base_url: 'https://api.x.ai/v1', + model_mapping: { + 'grok-latest': 'grok-4.3' + } + }, + extra: {}, + proxy_id: null, + concurrency: 1, + priority: 1, + rate_multiplier: 1, + status: 'active', + group_ids: [], + expires_at: null, + auto_pause_on_expired: false + } as any +} + function buildOpenAISetupTokenAccount() { return { ...buildAccount(), @@ -357,6 +383,35 @@ describe('EditAccountModal', () => { }) }) + it('loads and submits Grok OAuth model mapping edits', async () => { + const account = buildGrokOAuthAccount() + updateAccountMock.mockReset() + checkMixedChannelRiskMock.mockReset() + checkMixedChannelRiskMock.mockResolvedValue({ has_risk: false }) + updateAccountMock.mockResolvedValue(account) + + const wrapper = mountModal(account) + expect(wrapper.text()).toContain('Imagine Image') + expect(wrapper.text()).toContain('Imagine Video') + + const inputWithValue = (value: string) => { + const input = wrapper + .findAll('input') + .find((input) => (input.element as HTMLInputElement).value === value) + expect(input).toBeTruthy() + return input! + } + + await inputWithValue('grok-latest').setValue('grok') + await inputWithValue('grok-4.3').setValue('grok-build-0.1') + await wrapper.get('form#edit-account-form').trigger('submit.prevent') + + expect(updateAccountMock).toHaveBeenCalledTimes(1) + expect(updateAccountMock.mock.calls[0]?.[1]?.credentials?.model_mapping).toEqual({ + grok: 'grok-build-0.1' + }) + }) + it('only submits model mapping credentials when saving an OpenAI spark shadow account', async () => { authIsSimpleMode.value = false const account = buildOpenAISparkShadowAccount() diff --git a/frontend/src/composables/useModelWhitelist.ts b/frontend/src/composables/useModelWhitelist.ts index 43449b09ee..654e1bb973 100644 --- a/frontend/src/composables/useModelWhitelist.ts +++ b/frontend/src/composables/useModelWhitelist.ts @@ -141,7 +141,13 @@ const xaiModels = [ 'grok-latest', 'grok-build', 'grok-4.20-reasoning', - 'grok-4.20-non-reasoning' + 'grok-4.20-non-reasoning', + 'grok-imagine', + 'grok-imagine-image-quality', + 'grok-imagine-image', + 'grok-imagine-edit', + 'grok-imagine-video', + 'grok-imagine-video-1.5' ] // Cohere @@ -286,7 +292,10 @@ const grokPresetMappings = [ { label: 'Grok Latest', from: 'grok-latest', to: 'grok-4.3', color: 'bg-emerald-100 text-emerald-700 hover:bg-emerald-200 dark:bg-emerald-900/30 dark:text-emerald-400' }, { label: 'Build 0.1', from: 'grok-build', to: 'grok-build-0.1', color: 'bg-cyan-100 text-cyan-700 hover:bg-cyan-200 dark:bg-cyan-900/30 dark:text-cyan-400' }, { label: '4.20 Reasoning', from: 'grok-4.20-reasoning', to: 'grok-4.20-0309-reasoning', color: 'bg-indigo-100 text-indigo-700 hover:bg-indigo-200 dark:bg-indigo-900/30 dark:text-indigo-400' }, - { label: '4.20 Non Reasoning', from: 'grok-4.20-non-reasoning', to: 'grok-4.20-0309-non-reasoning', color: 'bg-violet-100 text-violet-700 hover:bg-violet-200 dark:bg-violet-900/30 dark:text-violet-400' } + { label: '4.20 Non Reasoning', from: 'grok-4.20-non-reasoning', to: 'grok-4.20-0309-non-reasoning', color: 'bg-violet-100 text-violet-700 hover:bg-violet-200 dark:bg-violet-900/30 dark:text-violet-400' }, + { label: 'Imagine Image', from: 'grok-imagine', to: 'grok-imagine-image-quality', color: 'bg-sky-100 text-sky-700 hover:bg-sky-200 dark:bg-sky-900/30 dark:text-sky-400' }, + { label: 'Imagine Edit', from: 'grok-imagine-edit', to: 'grok-imagine-edit', color: 'bg-rose-100 text-rose-700 hover:bg-rose-200 dark:bg-rose-900/30 dark:text-rose-400' }, + { label: 'Imagine Video', from: 'grok-imagine-video-1.5', to: 'grok-imagine-video-1.5', color: 'bg-amber-100 text-amber-700 hover:bg-amber-200 dark:bg-amber-900/30 dark:text-amber-400' } ] // Antigravity 预设映射(支持通配符) From f3a3a0869d099e64dcbad350cd30c71d87e97c81 Mon Sep 17 00:00:00 2001 From: jjaw Date: Fri, 3 Jul 2026 21:02:31 +0800 Subject: [PATCH 06/28] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E6=A7=BD=E4=BD=8D=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...eway_handler_warmup_intercept_unit_test.go | 1 + .../handler/gateway_helper_fastpath_test.go | 4 ++ .../handler/gateway_helper_hotpath_test.go | 4 ++ .../internal/repository/concurrency_cache.go | 50 +++++++++++++++++++ .../concurrency_cache_integration_test.go | 33 ++++++++++++ .../internal/service/concurrency_service.go | 19 +++---- .../service/concurrency_service_test.go | 4 ++ .../service/concurrency_slot_cleanup_test.go | 39 +++++++++++++++ .../service/gateway_multiplatform_test.go | 4 ++ backend/internal/testutil/stubs.go | 3 ++ 10 files changed, 148 insertions(+), 13 deletions(-) create mode 100644 backend/internal/service/concurrency_slot_cleanup_test.go diff --git a/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go b/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go index 09b20722c9..75e3c1aa8c 100644 --- a/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go +++ b/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go @@ -137,6 +137,7 @@ func (f *fakeConcurrencyCache) GetAccountConcurrencyBatch(_ context.Context, acc return result, nil } func (f *fakeConcurrencyCache) CleanupExpiredAccountSlots(context.Context, int64) error { return nil } +func (f *fakeConcurrencyCache) CleanupExpiredAccountSlotKeys(context.Context) error { return nil } func (f *fakeConcurrencyCache) CleanupStaleProcessSlots(context.Context, string) error { return nil } func newTestGatewayHandler(t *testing.T, group *service.Group, accounts []*service.Account) (*GatewayHandler, func()) { diff --git a/backend/internal/handler/gateway_helper_fastpath_test.go b/backend/internal/handler/gateway_helper_fastpath_test.go index c7c0fb6c9e..fecb9b071d 100644 --- a/backend/internal/handler/gateway_helper_fastpath_test.go +++ b/backend/internal/handler/gateway_helper_fastpath_test.go @@ -89,6 +89,10 @@ func (m *concurrencyCacheMock) CleanupExpiredAccountSlots(ctx context.Context, a return nil } +func (m *concurrencyCacheMock) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return nil +} + func (m *concurrencyCacheMock) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { return nil } diff --git a/backend/internal/handler/gateway_helper_hotpath_test.go b/backend/internal/handler/gateway_helper_hotpath_test.go index 65dc849683..fb17481f1c 100644 --- a/backend/internal/handler/gateway_helper_hotpath_test.go +++ b/backend/internal/handler/gateway_helper_hotpath_test.go @@ -140,6 +140,10 @@ func (s *helperConcurrencyCacheStub) CleanupExpiredAccountSlots(ctx context.Cont return nil } +func (s *helperConcurrencyCacheStub) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return nil +} + func (s *helperConcurrencyCacheStub) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { return nil } diff --git a/backend/internal/repository/concurrency_cache.go b/backend/internal/repository/concurrency_cache.go index 5e6f10062f..b530c30b90 100644 --- a/backend/internal/repository/concurrency_cache.go +++ b/backend/internal/repository/concurrency_cache.go @@ -174,6 +174,29 @@ var ( return 1 `) + // cleanupExpiredSlotKeysScript 批量清理实际存在的账号槽位键,避免后台任务从数据库加载全量账号。 + // KEYS = 有序集合键列表,ARGV[1] = TTL(秒)。 + cleanupExpiredSlotKeysScript = redis.NewScript(` + -- Redis 3.2-4.x compat: opt into effects replication so redis.call('TIME') + -- replicates correctly. No-op on Redis 5.0+ (effects replication is default). + redis.replicate_commands() + local ttl = tonumber(ARGV[1]) + local timeResult = redis.call('TIME') + local now = tonumber(timeResult[1]) + local expireBefore = now - ttl + local removed = 0 + for i = 1, #KEYS do + local key = KEYS[i] + removed = removed + redis.call('ZREMRANGEBYSCORE', key, '-inf', expireBefore) + if redis.call('ZCARD', key) == 0 then + redis.call('DEL', key) + else + redis.call('EXPIRE', key, ttl) + end + end + return removed + `) + // startupCleanupScript 清理非当前进程前缀的槽位成员。 // KEYS 是有序集合键列表,ARGV[1] 是当前进程前缀,ARGV[2] 是槽位 TTL。 // 遍历每个 KEYS[i],移除前缀不匹配的成员,清空后删 key,否则刷新 EXPIRE。 @@ -503,6 +526,10 @@ func (c *concurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, accou return err } +func (c *concurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return c.cleanupExpiredSlotKeysByPattern(ctx, accountSlotKeyPrefix+"*") +} + func (c *concurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { if activeRequestPrefix == "" { return nil @@ -527,6 +554,29 @@ func (c *concurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeR return nil } +// cleanupExpiredSlotKeysByPattern 扫描实际存在的账号槽位键并批量清理过期成员。 +func (c *concurrencyCache) cleanupExpiredSlotKeysByPattern(ctx context.Context, pattern string) error { + const scanCount = 200 + var cursor uint64 + for { + keys, nextCursor, err := c.rdb.Scan(ctx, cursor, pattern, scanCount).Result() + if err != nil { + return fmt.Errorf("scan %s: %w", pattern, err) + } + if len(keys) > 0 { + _, err := cleanupExpiredSlotKeysScript.Run(ctx, c.rdb, keys, c.slotTTLSeconds).Result() + if err != nil { + return fmt.Errorf("cleanup expired slots %s: %w", pattern, err) + } + } + cursor = nextCursor + if cursor == 0 { + break + } + } + return nil +} + // cleanupSlotsByPattern 扫描匹配 pattern 的有序集合键,批量调用 Lua 脚本清理非当前进程成员。 func (c *concurrencyCache) cleanupSlotsByPattern(ctx context.Context, pattern, activePrefix string) error { const scanCount = 200 diff --git a/backend/internal/repository/concurrency_cache_integration_test.go b/backend/internal/repository/concurrency_cache_integration_test.go index 5da94fc258..4aa172b252 100644 --- a/backend/internal/repository/concurrency_cache_integration_test.go +++ b/backend/internal/repository/concurrency_cache_integration_test.go @@ -435,6 +435,39 @@ func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlots_NoExpired() { require.Equal(s.T(), 2, cur) } +func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys() { + now := time.Now().Unix() + expiredTime := now - int64(testSlotTTL.Seconds()) - 10 + accountKeyWithFresh := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 301) + accountKeyExpiredOnly := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 302) + userKey := fmt.Sprintf("%s%d", userSlotKeyPrefix, 303) + + require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKeyWithFresh, + redis.Z{Score: float64(expiredTime), Member: "expired"}, + redis.Z{Score: float64(now), Member: "fresh"}, + ).Err()) + require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKeyExpiredOnly, + redis.Z{Score: float64(expiredTime), Member: "expired-only"}, + ).Err()) + require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userKey, + redis.Z{Score: float64(expiredTime), Member: "user-expired"}, + ).Err()) + + require.NoError(s.T(), s.cache.CleanupExpiredAccountSlotKeys(s.ctx)) + + accountMembers, err := s.rdb.ZRange(s.ctx, accountKeyWithFresh, 0, -1).Result() + require.NoError(s.T(), err) + require.Equal(s.T(), []string{"fresh"}, accountMembers) + + exists, err := s.rdb.Exists(s.ctx, accountKeyExpiredOnly).Result() + require.NoError(s.T(), err) + require.EqualValues(s.T(), 0, exists) + + userMembers, err := s.rdb.ZRange(s.ctx, userKey, 0, -1).Result() + require.NoError(s.T(), err) + require.Equal(s.T(), []string{"user-expired"}, userMembers) +} + func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_RemovesOldPrefixesAndWaitCounters() { accountID := int64(901) userID := int64(902) diff --git a/backend/internal/service/concurrency_service.go b/backend/internal/service/concurrency_service.go index 712fc1a749..50e0077b94 100644 --- a/backend/internal/service/concurrency_service.go +++ b/backend/internal/service/concurrency_service.go @@ -47,6 +47,7 @@ type ConcurrencyCache interface { // 清理过期槽位(后台任务) CleanupExpiredAccountSlots(ctx context.Context, accountID int64) error + CleanupExpiredAccountSlotKeys(ctx context.Context) error // 启动时清理旧进程遗留槽位与等待计数 CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error @@ -473,27 +474,19 @@ func (s *ConcurrencyService) CleanupExpiredAccountSlots(ctx context.Context, acc } // StartSlotCleanupWorker starts a background cleanup worker for expired account slots. -func (s *ConcurrencyService) StartSlotCleanupWorker(accountRepo AccountRepository, interval time.Duration) { - if s == nil || s.cache == nil || accountRepo == nil || interval <= 0 { +func (s *ConcurrencyService) StartSlotCleanupWorker(_ AccountRepository, interval time.Duration) { + if s == nil || s.cache == nil || interval <= 0 { return } runCleanup := func() { - listCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - accounts, err := accountRepo.ListSchedulable(listCtx) + cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + err := s.cache.CleanupExpiredAccountSlotKeys(cleanupCtx) cancel() if err != nil { - logger.LegacyPrintf("service.concurrency", "Warning: list schedulable accounts failed: %v", err) + logger.LegacyPrintf("service.concurrency", "Warning: cleanup expired account slots failed: %v", err) return } - for _, account := range accounts { - accountCtx, accountCancel := context.WithTimeout(context.Background(), 2*time.Second) - err := s.cache.CleanupExpiredAccountSlots(accountCtx, account.ID) - accountCancel() - if err != nil { - logger.LegacyPrintf("service.concurrency", "Warning: cleanup expired slots failed for account %d: %v", account.ID, err) - } - } } go func() { diff --git a/backend/internal/service/concurrency_service_test.go b/backend/internal/service/concurrency_service_test.go index 7d5f501dc5..bacad0245e 100644 --- a/backend/internal/service/concurrency_service_test.go +++ b/backend/internal/service/concurrency_service_test.go @@ -95,6 +95,10 @@ func (c *stubConcurrencyCacheForTest) CleanupExpiredAccountSlots(_ context.Conte return c.cleanupErr } +func (c *stubConcurrencyCacheForTest) CleanupExpiredAccountSlotKeys(_ context.Context) error { + return c.cleanupErr +} + func (c *stubConcurrencyCacheForTest) CleanupStaleProcessSlots(_ context.Context, _ string) error { return c.cleanupErr } diff --git a/backend/internal/service/concurrency_slot_cleanup_test.go b/backend/internal/service/concurrency_slot_cleanup_test.go new file mode 100644 index 0000000000..be26bb391b --- /dev/null +++ b/backend/internal/service/concurrency_slot_cleanup_test.go @@ -0,0 +1,39 @@ +package service + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +type slotCleanupCache struct { + ConcurrencyCache + calls atomic.Int64 +} + +func (c *slotCleanupCache) CleanupExpiredAccountSlotKeys(context.Context) error { + c.calls.Add(1) + return nil +} + +func TestStartSlotCleanupWorker_UsesCacheWideCleanupWithoutAccountRepo(t *testing.T) { + cache := &slotCleanupCache{} + svc := NewConcurrencyService(cache) + + svc.StartSlotCleanupWorker(nil, time.Hour) + + deadline := time.After(time.Second) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + if cache.calls.Load() > 0 { + return + } + select { + case <-deadline: + t.Fatal("cleanup worker did not call cache-wide account slot cleanup") + case <-ticker.C: + } + } +} diff --git a/backend/internal/service/gateway_multiplatform_test.go b/backend/internal/service/gateway_multiplatform_test.go index 291cea53b9..f843ba3e45 100644 --- a/backend/internal/service/gateway_multiplatform_test.go +++ b/backend/internal/service/gateway_multiplatform_test.go @@ -2097,6 +2097,10 @@ func (m *mockConcurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, a return nil } +func (m *mockConcurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return nil +} + func (m *mockConcurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { return nil } diff --git a/backend/internal/testutil/stubs.go b/backend/internal/testutil/stubs.go index bc572e1137..a0bd4bc669 100644 --- a/backend/internal/testutil/stubs.go +++ b/backend/internal/testutil/stubs.go @@ -76,6 +76,9 @@ func (c StubConcurrencyCache) GetAccountConcurrencyBatch(_ context.Context, acco func (c StubConcurrencyCache) CleanupExpiredAccountSlots(_ context.Context, _ int64) error { return nil } +func (c StubConcurrencyCache) CleanupExpiredAccountSlotKeys(_ context.Context) error { + return nil +} func (c StubConcurrencyCache) CleanupStaleProcessSlots(_ context.Context, _ string) error { return nil } From a1b2b32e08df7c3c878fce53061e380e59ae57a6 Mon Sep 17 00:00:00 2001 From: shaw Date: Fri, 3 Jul 2026 21:02:32 +0800 Subject: [PATCH 07/28] fix: prevent silent usage_logs drops under queue overflow (#3656) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 高并发下扣费成功但 usage_logs 被双层异步队列静默丢弃(单用户实测丢失 49%), 导致消费排行与对账缺口。根因是 6a685727 为避免溢出时单行 INSERT 踩踏数据库, 把三处"队列满"改成了立即终态丢弃。 改为有界阻塞背压,兼顾数据不丢与批量写入模式不变: - CreateBestEffort/createBatched 入队移除 default 立即丢弃分支, 队列满时阻塞等待(受 detached 15s ctx 期限约束) - writeUsageLogBestEffort 收到 dropped 不再直接放弃,统一走 repo.Create 同步兜底(仍经批处理器;ctx 耗尽时换新 detached 窗口), 重复写入由 ON CONFLICT DO NOTHING 幂等防护 - worker 池默认溢出策略 sample→sync(viper 默认 + 池常量), 溢出时提交方内联执行(提交点在响应写出后,不阻塞客户端), 显式配置 sample/drop 的部署不受影响 - 附带修复 ensure*Batcher 的存量数据竞争:channel nil 检查移入 sync.Once 正常负载零行为变化;突发时短暂等待替代永久丢失;数据库只见批量 INSERT。 --- backend/internal/config/config.go | 5 +- backend/internal/config/config_test.go | 4 +- backend/internal/repository/usage_log_repo.go | 27 +++++---- .../usage_log_repo_integration_test.go | 57 ++++++++++++++++--- .../service/gateway_record_usage_test.go | 8 ++- backend/internal/service/gateway_service.go | 13 ++++- .../service/usage_record_worker_pool.go | 9 +-- 7 files changed, 92 insertions(+), 31 deletions(-) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 18baa34881..b8b880eab0 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -1945,7 +1945,10 @@ func setDefaults() { viper.SetDefault("gateway.usage_record.worker_count", 128) viper.SetDefault("gateway.usage_record.queue_size", 16384) viper.SetDefault("gateway.usage_record.task_timeout_seconds", 5) - viper.SetDefault("gateway.usage_record.overflow_policy", UsageRecordOverflowPolicySample) + // 默认 sync:队列满时由提交方内联执行(提交点在响应写出之后,不阻塞客户端)。 + // sample/drop 会在溢出时静默丢弃计费任务,造成扣费与 usage_logs 对账缺口(issue #3656), + // 仅供显式配置的运维场景使用。 + viper.SetDefault("gateway.usage_record.overflow_policy", UsageRecordOverflowPolicySync) viper.SetDefault("gateway.usage_record.overflow_sample_percent", 10) viper.SetDefault("gateway.usage_record.auto_scale_enabled", true) viper.SetDefault("gateway.usage_record.auto_scale_min_workers", 128) diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go index bf7a327563..804155d1a9 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -1903,8 +1903,8 @@ func TestLoad_DefaultGatewayUsageRecordConfig(t *testing.T) { if cfg.Gateway.UsageRecord.TaskTimeoutSeconds != 5 { t.Fatalf("task_timeout_seconds = %d, want 5", cfg.Gateway.UsageRecord.TaskTimeoutSeconds) } - if cfg.Gateway.UsageRecord.OverflowPolicy != UsageRecordOverflowPolicySample { - t.Fatalf("overflow_policy = %s, want %s", cfg.Gateway.UsageRecord.OverflowPolicy, UsageRecordOverflowPolicySample) + if cfg.Gateway.UsageRecord.OverflowPolicy != UsageRecordOverflowPolicySync { + t.Fatalf("overflow_policy = %s, want %s", cfg.Gateway.UsageRecord.OverflowPolicy, UsageRecordOverflowPolicySync) } if cfg.Gateway.UsageRecord.OverflowSamplePercent != 10 { t.Fatalf("overflow_sample_percent = %d, want 10", cfg.Gateway.UsageRecord.OverflowSamplePercent) diff --git a/backend/internal/repository/usage_log_repo.go b/backend/internal/repository/usage_log_repo.go index 885e63f9fd..24c648b0a5 100644 --- a/backend/internal/repository/usage_log_repo.go +++ b/backend/internal/repository/usage_log_repo.go @@ -372,12 +372,13 @@ func (r *usageLogRepository) CreateBestEffort(ctx context.Context, log *service. } } + // 队列满时阻塞等待而非立即丢弃:批处理器持续排空队列,短暂等待即可入队。 + // 立即丢弃会造成“已扣费但无 usage_log”的永久数据缺口(issue #3656); + // 阻塞上限由调用方 ctx 期限约束,超时后由上层同步兜底。 select { case r.bestEffortBatchCh <- req: case <-ctx.Done(): return service.MarkUsageLogCreateDropped(ctx.Err()) - default: - return service.MarkUsageLogCreateDropped(errors.New("usage log best-effort queue full")) } select { @@ -493,12 +494,12 @@ func (r *usageLogRepository) createBatched(ctx context.Context, log *service.Usa resultCh: make(chan usageLogCreateResult, 1), } + // 队列满时阻塞等待而非立即报错:本路径是 best-effort 丢弃后的最后兜底, + // 立即失败会让日志永久丢失;阻塞上限由调用方 ctx 期限约束。 select { case r.createBatchCh <- req: case <-ctx.Done(): return false, service.MarkUsageLogCreateNotPersisted(ctx.Err()) - default: - return false, service.MarkUsageLogCreateNotPersisted(errors.New("usage log create batch queue full")) } select { @@ -520,22 +521,28 @@ func (r *usageLogRepository) createBatched(ctx context.Context, log *service.Usa } func (r *usageLogRepository) ensureCreateBatcher() { - if r == nil || r.db == nil || r.createBatchCh != nil { + if r == nil || r.db == nil { return } + // nil 检查必须在 Once 内部:在外层做无同步快路径读会与 Once 内的写构成数据竞争。 r.createBatchOnce.Do(func() { - r.createBatchCh = make(chan usageLogCreateRequest, usageLogCreateBatchQueueCap) - go r.runCreateBatcher(r.db) + if r.createBatchCh == nil { + r.createBatchCh = make(chan usageLogCreateRequest, usageLogCreateBatchQueueCap) + go r.runCreateBatcher(r.db) + } }) } func (r *usageLogRepository) ensureBestEffortBatcher() { - if r == nil || r.db == nil || r.bestEffortBatchCh != nil { + if r == nil || r.db == nil { return } + // 同 ensureCreateBatcher:nil 检查放在 Once 内部以避免数据竞争。 r.bestEffortBatchOnce.Do(func() { - r.bestEffortBatchCh = make(chan usageLogBestEffortRequest, usageLogBestEffortBatchQueueCap) - go r.runBestEffortBatcher(r.db) + if r.bestEffortBatchCh == nil { + r.bestEffortBatchCh = make(chan usageLogBestEffortRequest, usageLogBestEffortBatchQueueCap) + go r.runBestEffortBatcher(r.db) + } }) } diff --git a/backend/internal/repository/usage_log_repo_integration_test.go b/backend/internal/repository/usage_log_repo_integration_test.go index ed3050d89c..b43c6d56ef 100644 --- a/backend/internal/repository/usage_log_repo_integration_test.go +++ b/backend/internal/repository/usage_log_repo_integration_test.go @@ -288,21 +288,21 @@ func TestUsageLogRepositoryCreateBestEffort_BatchPathDuplicateRequestID(t *testi }, 3*time.Second, 20*time.Millisecond) } -func TestUsageLogRepositoryCreateBestEffort_QueueFullReturnsDropped(t *testing.T) { - ctx := context.Background() +func TestUsageLogRepositoryCreateBestEffort_QueueFullBlocksUntilCtxDeadline(t *testing.T) { + // 队列满时不再立即丢弃:阻塞等待入队,直到调用方 ctx 到期才标记 dropped(issue #3656)。 client := testEntClient(t) repo := newUsageLogRepositoryWithSQL(client, integrationDB) repo.bestEffortBatchCh = make(chan usageLogBestEffortRequest, 1) repo.bestEffortBatchCh <- usageLogBestEffortRequest{} - user := mustCreateUser(t, client, &service.User{Email: fmt.Sprintf("usage-best-effort-full-%d@example.com", time.Now().UnixNano())}) - apiKey := mustCreateApiKey(t, client, &service.APIKey{UserID: user.ID, Key: "sk-usage-best-effort-full-" + uuid.NewString(), Name: "k"}) - account := mustCreateAccount(t, client, &service.Account{Name: "acc-usage-best-effort-full-" + uuid.NewString()}) + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + start := time.Now() err := repo.CreateBestEffort(ctx, &service.UsageLog{ - UserID: user.ID, - APIKeyID: apiKey.ID, - AccountID: account.ID, + UserID: 1, + APIKeyID: 2, + AccountID: 3, RequestID: uuid.NewString(), Model: "claude-3", InputTokens: 10, @@ -314,6 +314,40 @@ func TestUsageLogRepositoryCreateBestEffort_QueueFullReturnsDropped(t *testing.T require.Error(t, err) require.True(t, service.IsUsageLogCreateDropped(err)) + require.GreaterOrEqual(t, time.Since(start), 150*time.Millisecond) +} + +func TestUsageLogRepositoryCreateBestEffort_QueueFullWaitsForDrain(t *testing.T) { + // 队列满但批处理器随后排空时,阻塞的入队应成功完成而非丢弃。 + client := testEntClient(t) + repo := newUsageLogRepositoryWithSQL(client, integrationDB) + repo.bestEffortBatchCh = make(chan usageLogBestEffortRequest, 1) + repo.bestEffortBatchCh <- usageLogBestEffortRequest{} + + go func() { + time.Sleep(100 * time.Millisecond) + <-repo.bestEffortBatchCh // 排空占位请求,为阻塞中的入队腾出空间 + req := <-repo.bestEffortBatchCh + sendUsageLogBestEffortResult(req.resultCh, nil) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + err := repo.CreateBestEffort(ctx, &service.UsageLog{ + UserID: 1, + APIKeyID: 2, + AccountID: 3, + RequestID: uuid.NewString(), + Model: "claude-3", + InputTokens: 10, + OutputTokens: 20, + TotalCost: 0.5, + ActualCost: 0.5, + CreatedAt: time.Now().UTC(), + }) + + require.NoError(t, err) } func TestUsageLogRepositoryCreate_BatchPathCanceledContextMarksNotPersisted(t *testing.T) { @@ -346,7 +380,7 @@ func TestUsageLogRepositoryCreate_BatchPathCanceledContextMarksNotPersisted(t *t } func TestUsageLogRepositoryCreate_BatchPathQueueFullMarksNotPersisted(t *testing.T) { - ctx := context.Background() + // 队列满时阻塞等待入队,直到调用方 ctx 到期才标记 not persisted(issue #3656)。 client := testEntClient(t) repo := newUsageLogRepositoryWithSQL(client, integrationDB) repo.createBatchCh = make(chan usageLogCreateRequest, 1) @@ -356,6 +390,10 @@ func TestUsageLogRepositoryCreate_BatchPathQueueFullMarksNotPersisted(t *testing apiKey := mustCreateApiKey(t, client, &service.APIKey{UserID: user.ID, Key: "sk-usage-create-full-" + uuid.NewString(), Name: "k"}) account := mustCreateAccount(t, client, &service.Account{Name: "acc-usage-create-full-" + uuid.NewString()}) + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + + start := time.Now() inserted, err := repo.Create(ctx, &service.UsageLog{ UserID: user.ID, APIKeyID: apiKey.ID, @@ -372,6 +410,7 @@ func TestUsageLogRepositoryCreate_BatchPathQueueFullMarksNotPersisted(t *testing require.False(t, inserted) require.Error(t, err) require.True(t, service.IsUsageLogCreateNotPersisted(err)) + require.GreaterOrEqual(t, time.Since(start), 150*time.Millisecond) } func TestUsageLogRepositoryCreate_BatchPathCanceledAfterQueueMarksNotPersisted(t *testing.T) { diff --git a/backend/internal/service/gateway_record_usage_test.go b/backend/internal/service/gateway_record_usage_test.go index c819eeca6e..2769251820 100644 --- a/backend/internal/service/gateway_record_usage_test.go +++ b/backend/internal/service/gateway_record_usage_test.go @@ -440,7 +440,9 @@ func TestGatewayServiceRecordUsage_GeneratesRequestIDWhenAllSourcesMissing(t *te require.Equal(t, billingRepo.lastCmd.RequestID, usageRepo.lastLog.RequestID) } -func TestGatewayServiceRecordUsage_DroppedUsageLogDoesNotSyncFallback(t *testing.T) { +func TestGatewayServiceRecordUsage_DroppedUsageLogFallsBackToSyncCreate(t *testing.T) { + // 计费成功后 best-effort 写入被丢弃(队列超时)时必须同步兜底, + // 否则出现“已扣费但无 usage_log”的对账缺口(issue #3656)。 usageRepo := &openAIRecordUsageBestEffortLogRepoStub{ bestEffortErr: MarkUsageLogCreateDropped(errors.New("usage log best-effort queue full")), } @@ -464,7 +466,9 @@ func TestGatewayServiceRecordUsage_DroppedUsageLogDoesNotSyncFallback(t *testing require.NoError(t, err) require.Equal(t, 1, usageRepo.bestEffortCalls) - require.Equal(t, 0, usageRepo.createCalls) + require.Equal(t, 1, usageRepo.createCalls) + // 兜底调用使用的 ctx 必须仍然存活,不能带着已死的 ctx 走过场。 + require.NoError(t, usageRepo.lastCtxErr) } func TestGatewayServiceRecordUsage_BillingErrorSkipsUsageLogWrite(t *testing.T) { diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 160a92a8e1..54035345d9 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -9473,10 +9473,17 @@ func writeUsageLogBestEffort(ctx context.Context, repo UsageLogRepository, usage if writer, ok := repo.(usageLogBestEffortWriter); ok { if err := writer.CreateBestEffort(usageCtx, usageLog); err != nil { logger.LegacyPrintf(logKey, "Create usage log failed: %v", err) - if IsUsageLogCreateDropped(err) { - return + // 计费已在此前完成,日志必须落库:dropped(批处理队列超时)同样走同步兜底, + // 否则会出现“已扣费但无 usage_log”的对账缺口(issue #3656)。 + // 重复写入由 usage_logs 的 ON CONFLICT (request_id, api_key_id) DO NOTHING 防护。 + fallbackCtx := usageCtx + if usageCtx.Err() != nil { + // usageCtx 已耗尽(best-effort 入队阻塞到期限):换新的 detached 窗口,避免兜底必然失败。 + var fallbackCancel context.CancelFunc + fallbackCtx, fallbackCancel = detachedBillingContext(context.Background()) + defer fallbackCancel() } - if _, syncErr := repo.Create(usageCtx, usageLog); syncErr != nil { + if _, syncErr := repo.Create(fallbackCtx, usageLog); syncErr != nil { logger.LegacyPrintf(logKey, "Create usage log sync fallback failed: %v", syncErr) } } diff --git a/backend/internal/service/usage_record_worker_pool.go b/backend/internal/service/usage_record_worker_pool.go index 5da0b89023..bb5ae452c8 100644 --- a/backend/internal/service/usage_record_worker_pool.go +++ b/backend/internal/service/usage_record_worker_pool.go @@ -15,10 +15,11 @@ import ( ) const ( - defaultUsageRecordWorkerCount = 128 - defaultUsageRecordQueueSize = 16384 - defaultUsageRecordTaskTimeoutSeconds = 5 - defaultUsageRecordOverflowPolicy = config.UsageRecordOverflowPolicySample + defaultUsageRecordWorkerCount = 128 + defaultUsageRecordQueueSize = 16384 + defaultUsageRecordTaskTimeoutSeconds = 5 + // 默认 sync:溢出时提交方内联执行,保证计费任务不被静默丢弃(issue #3656)。 + defaultUsageRecordOverflowPolicy = config.UsageRecordOverflowPolicySync defaultUsageRecordOverflowSampleRatio = 10 defaultUsageRecordAutoScaleEnabled = true defaultUsageRecordAutoScaleMinWorkers = 128 From f385cdceb78ba53108a8cf917277df10137be13b Mon Sep 17 00:00:00 2001 From: Anderson Date: Fri, 3 Jul 2026 22:36:39 +0800 Subject: [PATCH 08/28] feat: add Codex image tool strip policy --- .../service/codex_image_generation_bridge.go | 45 +++++++++++ .../service/openai_codex_transform.go | 39 ++++++--- .../service/openai_gateway_service.go | 21 ++++- .../openai_image_generation_controls_test.go | 39 +++++++++ .../internal/service/openai_ws_forwarder.go | 20 ++++- .../openai_ws_forwarder_ingress_test.go | 20 +++++ .../components/account/EditAccountModal.vue | 79 +++++++++++++++++++ .../__tests__/EditAccountModal.spec.ts | 35 ++++++++ frontend/src/i18n/locales/en.ts | 7 ++ frontend/src/i18n/locales/zh.ts | 6 ++ 10 files changed, 296 insertions(+), 15 deletions(-) diff --git a/backend/internal/service/codex_image_generation_bridge.go b/backend/internal/service/codex_image_generation_bridge.go index c7a894a792..68989d67dc 100644 --- a/backend/internal/service/codex_image_generation_bridge.go +++ b/backend/internal/service/codex_image_generation_bridge.go @@ -4,6 +4,13 @@ import "strings" const featureKeyCodexImageGenerationBridge = "codex_image_generation_bridge" +const ( + featureKeyCodexImageGenerationExplicitToolPolicy = "codex_image_generation_explicit_tool_policy" + + codexImageGenerationExplicitToolPolicyAllow = "allow" + codexImageGenerationExplicitToolPolicyStrip = "strip" +) + func boolOverridePtr(v bool) *bool { return &v } @@ -20,6 +27,27 @@ func boolOverrideFromMap(values map[string]any, keys ...string) *bool { return nil } +func stringOverrideFromMap(values map[string]any, keys ...string) (string, bool) { + if values == nil { + return "", false + } + for _, key := range keys { + if v, ok := values[key].(string); ok { + return v, true + } + } + return "", false +} + +func normalizeCodexImageGenerationExplicitToolPolicy(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case codexImageGenerationExplicitToolPolicyStrip, "remove", "drop": + return codexImageGenerationExplicitToolPolicyStrip + default: + return codexImageGenerationExplicitToolPolicyAllow + } +} + func platformBoolOverride(values map[string]any, key string, platform string) *bool { if values == nil { return nil @@ -62,3 +90,20 @@ func (a *Account) CodexImageGenerationBridgeOverride() *bool { openaiConfig, _ := a.Extra[PlatformOpenAI].(map[string]any) return boolOverrideFromMap(openaiConfig, featureKeyCodexImageGenerationBridge, "codex_image_generation_bridge_enabled") } + +// CodexImageGenerationExplicitToolPolicy returns the account-level policy for +// client-provided Codex /responses image_generation tools. Unknown or unset +// values default to allow to preserve existing behavior. +func (a *Account) CodexImageGenerationExplicitToolPolicy() string { + if a == nil || a.Platform != PlatformOpenAI || a.Extra == nil { + return codexImageGenerationExplicitToolPolicyAllow + } + if policy, ok := stringOverrideFromMap(a.Extra, featureKeyCodexImageGenerationExplicitToolPolicy); ok { + return normalizeCodexImageGenerationExplicitToolPolicy(policy) + } + openaiConfig, _ := a.Extra[PlatformOpenAI].(map[string]any) + if policy, ok := stringOverrideFromMap(openaiConfig, featureKeyCodexImageGenerationExplicitToolPolicy); ok { + return normalizeCodexImageGenerationExplicitToolPolicy(policy) + } + return codexImageGenerationExplicitToolPolicyAllow +} diff --git a/backend/internal/service/openai_codex_transform.go b/backend/internal/service/openai_codex_transform.go index 0ece5e44ff..c33ac97f0f 100644 --- a/backend/internal/service/openai_codex_transform.go +++ b/backend/internal/service/openai_codex_transform.go @@ -607,18 +607,21 @@ func hasOpenAIImageGenerationTool(reqBody map[string]any) bool { return false } -// stripCodexSparkImageGenerationTools removes image_generation tool entries from -// reqBody["tools"]. gpt-5.3-codex-spark rejects that tool upstream with HTTP 400 -// (invalid_request_error, param=tools), and Codex CLI advertises it by default, so -// it must be dropped for spark. When the tools list becomes empty the key is removed. -// Returns true when the body was modified. -func stripCodexSparkImageGenerationTools(reqBody map[string]any) bool { +func stripOpenAIImageGenerationTools(reqBody map[string]any) bool { rawTools, ok := reqBody["tools"] if !ok || rawTools == nil { + if openAIAnyToolChoiceSelectsImageGeneration(reqBody["tool_choice"]) { + delete(reqBody, "tool_choice") + return true + } return false } tools, ok := rawTools.([]any) if !ok { + if openAIAnyToolChoiceSelectsImageGeneration(reqBody["tool_choice"]) { + delete(reqBody, "tool_choice") + return true + } return false } filtered := make([]any, 0, len(tools)) @@ -631,17 +634,31 @@ func stripCodexSparkImageGenerationTools(reqBody map[string]any) bool { } filtered = append(filtered, rawTool) } - if !removed { + if !removed && !openAIAnyToolChoiceSelectsImageGeneration(reqBody["tool_choice"]) { return false } - if len(filtered) == 0 { - delete(reqBody, "tools") - } else { - reqBody["tools"] = filtered + if removed { + if len(filtered) == 0 { + delete(reqBody, "tools") + } else { + reqBody["tools"] = filtered + } + } + if openAIAnyToolChoiceSelectsImageGeneration(reqBody["tool_choice"]) { + delete(reqBody, "tool_choice") } return true } +// stripCodexSparkImageGenerationTools removes image_generation tool entries from +// reqBody["tools"]. gpt-5.3-codex-spark rejects that tool upstream with HTTP 400 +// (invalid_request_error, param=tools), and Codex CLI advertises it by default, so +// it must be dropped for spark. When the tools list becomes empty the key is removed. +// Returns true when the body was modified. +func stripCodexSparkImageGenerationTools(reqBody map[string]any) bool { + return stripOpenAIImageGenerationTools(reqBody) +} + func hasOpenAIInputImage(reqBody map[string]any) bool { if reqBody == nil { return false diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index cd84790b98..6bf04ef20f 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -2738,8 +2738,25 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if apiKey != nil { imageGenerationAllowed = GroupAllowsImageGeneration(apiKey.Group) } - codexImageGenerationBridgeEnabled := isCodexCLI && imageGenerationAllowed && s.isCodexImageGenerationBridgeEnabled(ctx, account, apiKey) - imageIntent := IsImageGenerationIntent(openAIResponsesEndpoint, reqModel, body) + codexImageGenerationExplicitToolPolicy := codexImageGenerationExplicitToolPolicyAllow + if isCodexCLI { + codexImageGenerationExplicitToolPolicy = account.CodexImageGenerationExplicitToolPolicy() + } + codexImageGenerationBridgeEnabled := isCodexCLI && imageGenerationAllowed && codexImageGenerationExplicitToolPolicy != codexImageGenerationExplicitToolPolicyStrip && s.isCodexImageGenerationBridgeEnabled(ctx, account, apiKey) + var imageIntent bool + if isCodexCLI && codexImageGenerationExplicitToolPolicy == codexImageGenerationExplicitToolPolicyStrip { + decoded, decodeErr := ensureReqBody() + if decodeErr != nil { + return nil, decodeErr + } + if stripOpenAIImageGenerationTools(decoded) { + markDecodedModified() + logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Stripped /responses image_generation tool for Codex client by account policy") + } + imageIntent = IsImageGenerationIntentMap(openAIResponsesEndpoint, reqModel, decoded) + } else { + imageIntent = IsImageGenerationIntent(openAIResponsesEndpoint, reqModel, body) + } if imageIntent && !imageGenerationAllowed { MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalFeatureGate) c.JSON(http.StatusForbidden, gin.H{"error": gin.H{"type": "permission_error", "message": ImageGenerationPermissionMessage()}}) diff --git a/backend/internal/service/openai_image_generation_controls_test.go b/backend/internal/service/openai_image_generation_controls_test.go index 6061a0bf9c..31edd36097 100644 --- a/backend/internal/service/openai_image_generation_controls_test.go +++ b/backend/internal/service/openai_image_generation_controls_test.go @@ -152,6 +152,45 @@ func TestOpenAIGatewayServiceForward_ExplicitImageToolWorksWithBridgeDisabled(t require.NotContains(t, instructions, "image_generation") } +func TestOpenAIGatewayServiceForward_AccountPolicyStripsExplicitImageTool(t *testing.T) { + gin.SetMode(gin.TestMode) + + upstream := &httpUpstreamRecorder{ + resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_stripped_image","model":"gpt-5.4","usage":{"input_tokens":2,"output_tokens":1}}`)), + }, + } + svc := newOpenAIImageGenerationControlTestService(upstream) + c, _ := newOpenAIImageGenerationControlTestContext(true, "codex_cli_rs/0.98.0") + account := newOpenAIImageGenerationControlTestAccount() + account.Extra = map[string]any{ + featureKeyCodexImageGenerationExplicitToolPolicy: codexImageGenerationExplicitToolPolicyStrip, + } + body := []byte(`{ + "model":"gpt-5.4", + "input":"draw", + "stream":false, + "tools":[ + {"type":"function","name":"shell","parameters":{"type":"object"}}, + {"type":"image_generation","format":"jpeg"} + ], + "tool_choice":{"type":"image_generation"} + }`) + + result, err := svc.Forward(context.Background(), c, account, body) + + require.NoError(t, err) + require.NotNil(t, result) + require.NotNil(t, upstream.lastReq) + require.False(t, gjson.GetBytes(upstream.lastBody, `tools.#(type=="image_generation")`).Exists()) + require.True(t, gjson.GetBytes(upstream.lastBody, `tools.#(type=="function")`).Exists()) + require.False(t, gjson.GetBytes(upstream.lastBody, "tool_choice").Exists()) + instructions := gjson.GetBytes(upstream.lastBody, "instructions").String() + require.NotContains(t, instructions, "image_generation") +} + func TestOpenAIGatewayServiceForward_ChannelBridgeOverrideEnablesCodexInjection(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_ws_forwarder.go b/backend/internal/service/openai_ws_forwarder.go index 067eeb6029..9261db5544 100644 --- a/backend/internal/service/openai_ws_forwarder.go +++ b/backend/internal/service/openai_ws_forwarder.go @@ -2448,11 +2448,15 @@ func stripCodexSparkImageGenerationToolFromRawPayload(payload []byte, model stri if !isCodexSparkModel(model) || !openAIRequestBodyHasImageGenerationTool(payload) { return payload, false, nil } + return stripOpenAIImageGenerationToolFromRawPayload(payload) +} + +func stripOpenAIImageGenerationToolFromRawPayload(payload []byte) ([]byte, bool, error) { payloadMap := make(map[string]any) if err := json.Unmarshal(payload, &payloadMap); err != nil { return payload, false, err } - if !stripCodexSparkImageGenerationTools(payloadMap) { + if !stripOpenAIImageGenerationTools(payloadMap) { return payload, false, nil } rebuilt, err := json.Marshal(payloadMap) @@ -2671,7 +2675,11 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( } apiKey := getAPIKeyFromContext(c) imageGenerationAllowed := GroupAllowsImageGeneration(apiKeyGroup(apiKey)) - codexBridgeEnabled := isCodexCLI && imageGenerationAllowed && s.isCodexImageGenerationBridgeEnabled(ctx, account, apiKey) + codexImageGenerationExplicitToolPolicy := codexImageGenerationExplicitToolPolicyAllow + if isCodexCLI { + codexImageGenerationExplicitToolPolicy = account.CodexImageGenerationExplicitToolPolicy() + } + codexBridgeEnabled := isCodexCLI && imageGenerationAllowed && codexImageGenerationExplicitToolPolicy != codexImageGenerationExplicitToolPolicyStrip && s.isCodexImageGenerationBridgeEnabled(ctx, account, apiKey) if codexBridgeEnabled { payloadMap := make(map[string]any) if err := json.Unmarshal(normalized, &payloadMap); err != nil { @@ -2709,6 +2717,14 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( } normalized = next } + if isCodexCLI && codexImageGenerationExplicitToolPolicy == codexImageGenerationExplicitToolPolicyStrip { + if stripped, changed, stripErr := stripOpenAIImageGenerationToolFromRawPayload(normalized); stripErr != nil { + return openAIWSClientPayload{}, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket request payload", stripErr) + } else if changed { + normalized = stripped + logOpenAIWSModeInfo("ingress_ws_codex_image_tool_stripped_by_policy account_id=%d", account.ID) + } + } if stripped, changed, stripErr := stripCodexSparkImageGenerationToolFromRawPayload(normalized, upstreamModel); stripErr != nil { return openAIWSClientPayload{}, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket request payload", stripErr) } else if changed { diff --git a/backend/internal/service/openai_ws_forwarder_ingress_test.go b/backend/internal/service/openai_ws_forwarder_ingress_test.go index 0d19a189b0..ca7c36aaa7 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress_test.go +++ b/backend/internal/service/openai_ws_forwarder_ingress_test.go @@ -169,6 +169,26 @@ func TestStripCodexSparkImageGenerationToolFromRawPayload(t *testing.T) { }) } +func TestStripOpenAIImageGenerationToolFromRawPayload(t *testing.T) { + payload := []byte(`{ + "type":"response.create", + "model":"gpt-5.4", + "tools":[ + {"type":"function","name":"shell"}, + {"type":"image_generation","output_format":"png"} + ], + "tool_choice":{"type":"image_generation"} + }`) + + updated, changed, err := stripOpenAIImageGenerationToolFromRawPayload(payload) + + require.NoError(t, err) + require.True(t, changed) + require.False(t, gjson.GetBytes(updated, `tools.#(type=="image_generation")`).Exists()) + require.True(t, gjson.GetBytes(updated, `tools.#(type=="function")`).Exists()) + require.False(t, gjson.GetBytes(updated, "tool_choice").Exists()) +} + func TestAlignStoreDisabledPreviousResponseID(t *testing.T) { t.Parallel() diff --git a/frontend/src/components/account/EditAccountModal.vue b/frontend/src/components/account/EditAccountModal.vue index b5f97753d8..57723d4903 100644 --- a/frontend/src/components/account/EditAccountModal.vue +++ b/frontend/src/components/account/EditAccountModal.vue @@ -1430,6 +1430,58 @@
+ +
+
+
+
+ +
+
+ +

+ {{ t('admin.accounts.openai.codexImageToolPolicyDesc') }} +

+
+
+
+
+ +
+
+
+
+
('inherit') +type CodexImageToolPolicyMode = 'allow' | 'strip' +const codexImageToolPolicyMode = ref('allow') type AnthropicAPIKeyAuthScheme = 'x_api_key' | 'authorization_bearer' const anthropicPassthroughEnabled = ref(false) const anthropicAPIKeyAuthScheme = ref('x_api_key') @@ -2703,6 +2757,22 @@ const codexImageGenerationBridgeOptions = computed>(() => [ + { + value: 'allow', + label: t('admin.accounts.openai.codexImageToolPolicyAllow'), + description: t('admin.accounts.openai.codexImageToolPolicyAllowDesc') + }, + { + value: 'strip', + label: t('admin.accounts.openai.codexImageToolPolicyStrip'), + description: t('admin.accounts.openai.codexImageToolPolicyStripDesc') + } +]) const codexImageGenerationBridgeBadgeLabel = computed(() => { switch (codexImageGenerationBridgeMode.value) { case 'enabled': @@ -3036,6 +3106,7 @@ const syncFormFromAccount = (newAccount: Account | null) => { codexCLIOnlyEnabled.value = false codexCLIOnlyAppServerEnabled.value = false codexImageGenerationBridgeMode.value = 'inherit' + codexImageToolPolicyMode.value = 'allow' anthropicPassthroughEnabled.value = false anthropicAPIKeyAuthScheme.value = 'x_api_key' webSearchEmulationMode.value = 'default' @@ -3059,6 +3130,9 @@ const syncFormFromAccount = (newAccount: Account | null) => { } else if (codexImageGenerationBridgeValue === false) { codexImageGenerationBridgeMode.value = 'disabled' } + codexImageToolPolicyMode.value = extra?.codex_image_generation_explicit_tool_policy === 'strip' + ? 'strip' + : 'allow' openaiOAuthResponsesWebSocketV2Mode.value = resolveOpenAIWSModeFromExtra(extra, { modeKey: 'openai_oauth_responses_websockets_v2_mode', enabledKey: 'openai_oauth_responses_websockets_v2_enabled', @@ -4199,6 +4273,11 @@ const handleSubmit = async () => { } else { newExtra.codex_image_generation_bridge = codexImageGenerationBridgeMode.value === 'enabled' } + if (codexImageToolPolicyMode.value === 'strip') { + newExtra.codex_image_generation_explicit_tool_policy = 'strip' + } else { + delete newExtra.codex_image_generation_explicit_tool_policy + } if (props.account.type === 'oauth' || props.account.type === 'setup-token') { if (codexCLIOnlyEnabled.value) { diff --git a/frontend/src/components/account/__tests__/EditAccountModal.spec.ts b/frontend/src/components/account/__tests__/EditAccountModal.spec.ts index 381d512ee7..c3646a8e51 100644 --- a/frontend/src/components/account/__tests__/EditAccountModal.spec.ts +++ b/frontend/src/components/account/__tests__/EditAccountModal.spec.ts @@ -577,6 +577,41 @@ describe('EditAccountModal', () => { expect(updateAccountMock.mock.calls[0]?.[1]?.extra).not.toHaveProperty('codex_image_generation_bridge_enabled') }) + it('submits account-level Codex image_generation tool strip policy', async () => { + const account = buildAccount() + updateAccountMock.mockReset() + checkMixedChannelRiskMock.mockReset() + checkMixedChannelRiskMock.mockResolvedValue({ has_risk: false }) + updateAccountMock.mockResolvedValue(account) + + const wrapper = mountModal(account) + + await wrapper.get('button[data-testid="codex-image-tool-policy-strip"]').trigger('click') + await wrapper.get('form#edit-account-form').trigger('submit.prevent') + + expect(updateAccountMock).toHaveBeenCalledTimes(1) + expect(updateAccountMock.mock.calls[0]?.[1]?.extra?.codex_image_generation_explicit_tool_policy).toBe('strip') + }) + + it('keeps account-level Codex image_generation tool policy unset when allowing', async () => { + const account = buildAccount() + account.extra = { + codex_image_generation_explicit_tool_policy: 'strip' + } + updateAccountMock.mockReset() + checkMixedChannelRiskMock.mockReset() + checkMixedChannelRiskMock.mockResolvedValue({ has_risk: false }) + updateAccountMock.mockResolvedValue(account) + + const wrapper = mountModal(account) + + await wrapper.get('button[data-testid="codex-image-tool-policy-allow"]').trigger('click') + await wrapper.get('form#edit-account-form').trigger('submit.prevent') + + expect(updateAccountMock).toHaveBeenCalledTimes(1) + expect(updateAccountMock.mock.calls[0]?.[1]?.extra).not.toHaveProperty('codex_image_generation_explicit_tool_policy') + }) + it('setup-token account can select and submit OAuth WS mode', async () => { const account = buildOpenAISetupTokenAccount() updateAccountMock.mockReset() diff --git a/frontend/src/i18n/locales/en.ts b/frontend/src/i18n/locales/en.ts index 831b7d7eae..4b6ec9f35f 100644 --- a/frontend/src/i18n/locales/en.ts +++ b/frontend/src/i18n/locales/en.ts @@ -3569,6 +3569,13 @@ export default { codexImageGenerationBridgeBadgeInherit: 'Channel policy', codexImageGenerationBridgeBadgeEnabled: 'Account on', codexImageGenerationBridgeBadgeDisabled: 'Account off', + codexImageToolPolicy: 'Codex image tool forwarding', + codexImageToolPolicyDesc: + 'Only applies to client-provided image_generation tools in Codex /responses text requests; standalone image-generation endpoints are unaffected.', + codexImageToolPolicyAllow: 'Allow', + codexImageToolPolicyAllowDesc: 'Default behavior. Keep client-provided image_generation tools.', + codexImageToolPolicyStrip: 'Remove', + codexImageToolPolicyStripDesc: 'Delete image_generation tools and matching tool_choice before forwarding.', compactMode: 'Compact mode', compactModeDesc: 'Controls how this account participates in /responses/compact routing. Auto follows probe results, Force On always allows, Force Off always excludes.', diff --git a/frontend/src/i18n/locales/zh.ts b/frontend/src/i18n/locales/zh.ts index 64c62f39f8..9c76037d9d 100644 --- a/frontend/src/i18n/locales/zh.ts +++ b/frontend/src/i18n/locales/zh.ts @@ -3742,6 +3742,12 @@ export default { codexImageGenerationBridgeBadgeInherit: '渠道策略', codexImageGenerationBridgeBadgeEnabled: '账号开启', codexImageGenerationBridgeBadgeDisabled: '账号关闭', + codexImageToolPolicy: 'Codex 图片工具传递策略', + codexImageToolPolicyDesc: '仅作用于 Codex /responses 文本端点中客户端显式携带的 image_generation tool;不影响独立图片生成接口。', + codexImageToolPolicyAllow: '放行', + codexImageToolPolicyAllowDesc: '默认行为,保留客户端显式发送的 image_generation tool。', + codexImageToolPolicyStrip: '移除', + codexImageToolPolicyStripDesc: '转发前删除 image_generation tool 和指向它的 tool_choice。', compactMode: 'Compact 模式', compactModeDesc: '控制本账号在 /responses/compact 调度中的参与方式。Auto 跟随探测结果,Force On 强制允许,Force Off 强制排除。', From ebbdc70311781325d9db2ece7ef1936b487f6053 Mon Sep 17 00:00:00 2001 From: DaydreamCoding <22166516+DaydreamCoding@users.noreply.github.com> Date: Fri, 3 Jul 2026 23:02:27 +0800 Subject: [PATCH 09/28] =?UTF-8?q?feat(usage):=20=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E8=AF=B7=E6=B1=82=E5=AF=B9=E9=BD=90=E7=94=A8=E9=87=8F=E6=98=8E?= =?UTF-8?q?=E7=BB=86(UI/=E6=8E=92=E5=BA=8F/=E7=AD=9B=E9=80=89/=E5=88=97?= =?UTF-8?q?=E8=AE=BE=E7=BD=AE)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 错误请求列表(/admin/usage 错误 tab、Ops 弹窗、用户端 /usage 错误 tab) 对齐用量明细的交互与信息密度。Squash of: - feat(usage): 错误请求全面对齐用量明细(UI/列序/排序/筛选/列设置/新列) - refactor(usage): 错误表提取共享徽章工具与 IP 批量工具条,后端排序解析归并 SetSort - feat(usage): /admin/usage 错误请求新增分类过滤 - fix(ops): 错误列表 phase=upstream 过滤生效,守卫豁免改为显式 opt-in - fix(ops): 错误列表用户列回退显示已删除 KEY 所有者 - fix(usage): 错误请求状态码排序对齐展示/过滤,筛选项改固定常用码 --- backend/internal/handler/admin/ops_handler.go | 49 +- backend/internal/handler/usage_handler.go | 3 + .../repository/ops_error_where_test.go | 17 +- backend/internal/repository/ops_repo.go | 60 ++- backend/internal/service/ops_models.go | 42 +- backend/internal/service/ops_service.go | 8 +- .../service/ops_service_user_error_test.go | 18 +- backend/internal/service/ops_user_error.go | 21 +- .../internal/service/ops_user_error_test.go | 20 +- frontend/src/__tests__/setup.ts | 14 + frontend/src/api/admin/ops.ts | 18 +- frontend/src/api/admin/usage.ts | 4 + .../components/admin/usage/UsageFilters.vue | 73 ++- frontend/src/components/common/DataTable.vue | 7 + .../components/common/IpGeoBatchToolbar.vue | 59 +++ .../user/UserErrorRequestsTable.vue | 281 +++++----- frontend/src/i18n/locales/en.ts | 3 +- frontend/src/i18n/locales/zh.ts | 3 +- frontend/src/types/index.ts | 8 + frontend/src/utils/errorBadges.ts | 61 +++ frontend/src/utils/errorCategory.ts | 31 ++ frontend/src/views/admin/UsageView.vue | 97 +++- .../ops/components/OpsErrorDetailsModal.vue | 15 +- .../admin/ops/components/OpsErrorLogTable.vue | 497 +++++++++--------- frontend/src/views/user/UsageView.vue | 158 +++++- 25 files changed, 1108 insertions(+), 459 deletions(-) create mode 100644 frontend/src/components/common/IpGeoBatchToolbar.vue create mode 100644 frontend/src/utils/errorBadges.ts create mode 100644 frontend/src/utils/errorCategory.ts diff --git a/backend/internal/handler/admin/ops_handler.go b/backend/internal/handler/admin/ops_handler.go index b9558b97b1..e820aef0c6 100644 --- a/backend/internal/handler/admin/ops_handler.go +++ b/backend/internal/handler/admin/ops_handler.go @@ -73,6 +73,13 @@ func NewOpsHandler(opsService *service.OpsService) *OpsHandler { } // GetErrorLogs lists ops error logs. +// applyOpsErrorSortParams reads sort_by/sort_order query params into the filter. +// Column whitelist and order normalization live in the repository; unknown +// values degrade to the default (created_at DESC), mirroring the usage list. +func applyOpsErrorSortParams(c *gin.Context, filter *service.OpsErrorLogFilter) { + filter.SetSort(c.Query("sort_by"), c.Query("sort_order")) +} + // GET /api/v1/admin/ops/errors func (h *OpsHandler) GetErrorLogs(c *gin.Context) { if h.opsService == nil { @@ -114,10 +121,17 @@ func (h *OpsHandler) GetErrorLogs(c *gin.Context) { // buildOpsErrorLogsWhere 以 COALESCE(requested_model, model) 比对。 filter.Model = strings.TrimSpace(c.Query("model")) - // Force request errors: client-visible status >= 400. - // buildOpsErrorLogsWhere already applies this for non-upstream phase. - if strings.EqualFold(strings.TrimSpace(filter.Phase), "upstream") { - filter.Phase = "" + // 请求错误语义:client-visible status>=400 守卫恒生效(未设 + // IncludeRecoveredUpstream 时 phase=upstream 不再绕过守卫),故 + // phase=upstream 作为普通过滤条件保留——此前这里清空该值,导致 + // 错误类型下拉选「上游」等于不过滤。 + + // 分类(用户侧粗分类码)→ phase/type ANY 条件,与用户端 /usage/errors 同一映射; + // 未知分类返回空切片 = 不过滤。与 phase 参数可同时设置(AND 语义)。 + if cat := strings.TrimSpace(c.Query("category")); cat != "" { + phases, types := service.CategoryToFilter(cat) + filter.ErrorPhasesAny = phases + filter.ErrorTypesAny = types } if platform := strings.TrimSpace(c.Query("platform")); platform != "" { @@ -187,6 +201,8 @@ func (h *OpsHandler) GetErrorLogs(c *gin.Context) { filter.StatusCodes = out } + applyOpsErrorSortParams(c, filter) + result, err := h.opsService.GetErrorLogs(c.Request.Context(), filter) if err != nil { response.ErrorFrom(c, err) @@ -234,10 +250,17 @@ func (h *OpsHandler) ListRequestErrors(c *gin.Context) { // buildOpsErrorLogsWhere 以 COALESCE(requested_model, model) 比对。 filter.Model = strings.TrimSpace(c.Query("model")) - // Force request errors: client-visible status >= 400. - // buildOpsErrorLogsWhere already applies this for non-upstream phase. - if strings.EqualFold(strings.TrimSpace(filter.Phase), "upstream") { - filter.Phase = "" + // 请求错误语义:client-visible status>=400 守卫恒生效(未设 + // IncludeRecoveredUpstream 时 phase=upstream 不再绕过守卫),故 + // phase=upstream 作为普通过滤条件保留——此前这里清空该值,导致 + // 错误类型下拉选「上游」等于不过滤。 + + // 分类(用户侧粗分类码)→ phase/type ANY 条件,与用户端 /usage/errors 同一映射; + // 未知分类返回空切片 = 不过滤。与 phase 参数可同时设置(AND 语义)。 + if cat := strings.TrimSpace(c.Query("category")); cat != "" { + phases, types := service.CategoryToFilter(cat) + filter.ErrorPhasesAny = phases + filter.ErrorTypesAny = types } if platform := strings.TrimSpace(c.Query("platform")); platform != "" { @@ -291,6 +314,8 @@ func (h *OpsHandler) ListRequestErrors(c *gin.Context) { filter.StatusCodes = out } + applyOpsErrorSortParams(c, filter) + result, err := h.opsService.GetErrorLogs(c.Request.Context(), filter) if err != nil { response.ErrorFrom(c, err) @@ -362,6 +387,8 @@ func (h *OpsHandler) ListRequestErrorUpstreamErrors(c *gin.Context) { } filter.View = "all" filter.Phase = "upstream" + // 上游错误列表需含 status<400 的 recovered 行,显式豁免客户端可见守卫。 + filter.IncludeRecoveredUpstream = true filter.Owner = "provider" filter.Source = strings.TrimSpace(c.Query("error_source")) filter.Query = strings.TrimSpace(c.Query("q")) @@ -377,6 +404,8 @@ func (h *OpsHandler) ListRequestErrorUpstreamErrors(c *gin.Context) { filter.ClientRequestID = clientRequestID } + applyOpsErrorSortParams(c, filter) + result, err := h.opsService.GetErrorLogs(c.Request.Context(), filter) if err != nil { response.ErrorFrom(c, err) @@ -442,6 +471,8 @@ func (h *OpsHandler) ListUpstreamErrors(c *gin.Context) { filter.View = parseOpsViewParam(c) filter.Phase = "upstream" + // 上游错误列表需含 status<400 的 recovered 行,显式豁免客户端可见守卫。 + filter.IncludeRecoveredUpstream = true filter.Owner = "provider" filter.Source = strings.TrimSpace(c.Query("error_source")) filter.Query = strings.TrimSpace(c.Query("q")) @@ -497,6 +528,8 @@ func (h *OpsHandler) ListUpstreamErrors(c *gin.Context) { filter.StatusCodes = out } + applyOpsErrorSortParams(c, filter) + result, err := h.opsService.GetErrorLogs(c.Request.Context(), filter) if err != nil { response.ErrorFrom(c, err) diff --git a/backend/internal/handler/usage_handler.go b/backend/internal/handler/usage_handler.go index 9d0f1d8fac..be6dc917bb 100644 --- a/backend/internal/handler/usage_handler.go +++ b/backend/internal/handler/usage_handler.go @@ -322,6 +322,9 @@ func (h *UsageHandler) ListErrors(c *gin.Context) { filter.ErrorTypesAny = types } + // 排序对齐用量明细:列白名单与方向归一在 repo 层,非法值回退 created_at DESC。 + filter.SetSort(c.Query("sort_by"), c.Query("sort_order")) + result, err := h.opsService.ListUserErrorRequests(c.Request.Context(), subject.UserID, filter) if err != nil { response.ErrorFrom(c, err) diff --git a/backend/internal/repository/ops_error_where_test.go b/backend/internal/repository/ops_error_where_test.go index 5b9d7ab1c3..c997865ba4 100644 --- a/backend/internal/repository/ops_error_where_test.go +++ b/backend/internal/repository/ops_error_where_test.go @@ -85,10 +85,21 @@ func TestBuildOpsErrorLogsWhere_CyberPolicyStatusExemption(t *testing.T) { t.Fatalf("default filter must still include the status >= 400 guard for non-cyber rows\nfull: %s", where) } - // phase=upstream skips the status guard entirely — exemption is irrelevant there. + // phase=upstream WITHOUT the recovered-upstream opt-in keeps the status guard: + // request-error list endpoints filter by phase=upstream as a plain condition. whereUpstream, _ := buildOpsErrorLogsWhere(&service.OpsErrorLogFilter{Phase: "upstream"}) - if strings.Contains(whereUpstream, "status_code") { - t.Fatalf("upstream phase filter must not add any status_code clause\nfull: %s", whereUpstream) + if !strings.Contains(whereUpstream, "COALESCE(e.status_code, 0) >= 400") { + t.Fatalf("upstream phase without IncludeRecoveredUpstream must keep the status guard\nfull: %s", whereUpstream) + } + if !strings.Contains(whereUpstream, "e.error_phase = $") { + t.Fatalf("upstream phase filter must emit the error_phase condition\nfull: %s", whereUpstream) + } + + // phase=upstream WITH IncludeRecoveredUpstream (ops 上游列表) skips the guard, + // exposing recovered (<400) upstream rows. + whereRecovered, _ := buildOpsErrorLogsWhere(&service.OpsErrorLogFilter{Phase: "upstream", IncludeRecoveredUpstream: true}) + if strings.Contains(whereRecovered, "status_code") { + t.Fatalf("upstream phase with IncludeRecoveredUpstream must not add any status_code clause\nfull: %s", whereRecovered) } } diff --git a/backend/internal/repository/ops_repo.go b/backend/internal/repository/ops_repo.go index 9923c08d99..2129a451c4 100644 --- a/backend/internal/repository/ops_repo.go +++ b/backend/internal/repository/ops_repo.go @@ -177,6 +177,37 @@ func opsInsertErrorLogArgs(input *service.OpsInsertErrorLogInput) []any { } } +// opsErrorLogsOrderBy builds the ORDER BY clause from a whitelist, mirroring +// usageLogOrderBy semantics. Unknown SortBy falls back to created_at; e.id is +// always appended as tiebreaker for stable pagination. +func opsErrorLogsOrderBy(filter *service.OpsErrorLogFilter) string { + sortBy := "" + sortOrder := "" + if filter != nil { + sortBy = strings.ToLower(strings.TrimSpace(filter.SortBy)) + sortOrder = strings.ToLower(strings.TrimSpace(filter.SortOrder)) + } + + var column string + switch sortBy { + case "model": + column = "COALESCE(NULLIF(TRIM(e.requested_model), ''), e.model)" + case "status_code": + // 与展示列/过滤保持同义:列表展示 COALESCE(upstream_status_code, status_code, 0), + // status_code 过滤也用同一表达式,故排序必须一致——否则 recovered upstream 行 + //(status_code<400 但展示上游 5xx)排序键与显示值/分页切分不符。 + column = "COALESCE(e.upstream_status_code, e.status_code, 0)" + default: + column = "e.created_at" + } + + dir := "DESC" + if sortOrder == "asc" { + dir = "ASC" + } + return fmt.Sprintf("%s %s, e.id %s", column, dir, dir) +} + func (r *opsRepository) ListErrorLogs(ctx context.Context, filter *service.OpsErrorLogFilter) (*service.OpsErrorLogList, error) { if r == nil || r.db == nil { return nil, fmt.Errorf("nil ops repository") @@ -233,25 +264,29 @@ SELECT COALESCE(a.name, ''), e.group_id, COALESCE(g.name, ''), - CASE WHEN e.client_ip IS NULL THEN NULL ELSE e.client_ip::text END, + CASE WHEN e.client_ip IS NULL THEN NULL ELSE host(e.client_ip) END, COALESCE(e.request_path, ''), e.stream, COALESCE(e.inbound_endpoint, ''), COALESCE(e.upstream_endpoint, ''), COALESCE(e.requested_model, ''), COALESCE(e.upstream_model, ''), + COALESCE(e.user_agent, ''), e.request_type, COALESCE(ak.name, ''), ak.deleted_at, - COALESCE(e.deleted_key_name, '') + COALESCE(e.deleted_key_name, ''), + e.deleted_key_owner_user_id, + COALESCE(du.email, '') FROM ops_error_logs e LEFT JOIN accounts a ON e.account_id = a.id LEFT JOIN groups g ON e.group_id = g.id LEFT JOIN users u ON e.user_id = u.id LEFT JOIN users u2 ON e.resolved_by_user_id = u2.id +LEFT JOIN users du ON e.deleted_key_owner_user_id = du.id LEFT JOIN api_keys ak ON ak.id = e.api_key_id ` + where + ` -ORDER BY e.created_at DESC +ORDER BY ` + opsErrorLogsOrderBy(filter) + ` LIMIT $` + itoa(len(args)+1) + ` OFFSET $` + itoa(len(args)+2) rows, err := r.db.QueryContext(ctx, selectSQL, argsWithLimit...) @@ -279,6 +314,8 @@ LIMIT $` + itoa(len(args)+1) + ` OFFSET $` + itoa(len(args)+2) var apiKeyName string var apiKeyDeletedAt sql.NullTime var deletedKeyName string + var deletedKeyOwnerID sql.NullInt64 + var deletedKeyOwnerEmail string if err := rows.Scan( &item.ID, &item.CreatedAt, @@ -311,10 +348,13 @@ LIMIT $` + itoa(len(args)+1) + ` OFFSET $` + itoa(len(args)+2) &item.UpstreamEndpoint, &item.RequestedModel, &item.UpstreamModel, + &item.UserAgent, &requestType, &apiKeyName, &apiKeyDeletedAt, &deletedKeyName, + &deletedKeyOwnerID, + &deletedKeyOwnerEmail, ); err != nil { return nil, err } @@ -364,6 +404,12 @@ LIMIT $` + itoa(len(args)+1) + ` OFFSET $` + itoa(len(args)+2) } // 已删除:ak.deleted_at 非空(软删),或仅命中 deleted_key_name 兜底。 item.APIKeyDeleted = apiKeyDeletedAt.Valid || (apiKeyName == "" && deletedKeyName != "") + // 已删除 KEY 所有者快照:认证失败行 user_id 为空,列表用户列以此回退。 + if deletedKeyOwnerID.Valid { + v := deletedKeyOwnerID.Int64 + item.DeletedKeyOwnerUserID = &v + item.DeletedKeyOwnerEmail = deletedKeyOwnerEmail + } out = append(out, &item) } if err := rows.Err(); err != nil { @@ -417,7 +463,7 @@ SELECT COALESCE(a.name, ''), e.group_id, COALESCE(g.name, ''), - CASE WHEN e.client_ip IS NULL THEN NULL ELSE e.client_ip::text END, + CASE WHEN e.client_ip IS NULL THEN NULL ELSE host(e.client_ip) END, COALESCE(e.request_path, ''), e.stream, COALESCE(e.inbound_endpoint, ''), @@ -927,12 +973,14 @@ func buildOpsErrorLogsWhere(filter *service.OpsErrorLogFilter) (string, []any) { if filter != nil { resolvedFilter = filter.Resolved } - // Keep list endpoints scoped to client errors unless explicitly filtering upstream phase. + // Keep list endpoints scoped to client errors unless the caller explicitly opts + // into recovered upstream rows (Phase=="upstream" + IncludeRecoveredUpstream, + // ops 专用上游列表)。请求错误语义的端点即便过滤 phase=upstream 也保留该守卫。 // cyber_policy is exempt from the status >= 400 guard: streaming cyber hits arrive with // status 200 (the SSE stream opened successfully before upstream returned response.failed), // but they are always client-visible blocked requests that belong in admin + user error // lists. Without the exemption the entire streaming-path cyber sink would be invisible. - if phaseFilter != "upstream" { + if phaseFilter != "upstream" || filter == nil || !filter.IncludeRecoveredUpstream { clauses = append(clauses, "(COALESCE(e.status_code, 0) >= 400 OR e.error_type = 'cyber_policy')") } diff --git a/backend/internal/service/ops_models.go b/backend/internal/service/ops_models.go index 4fc6a9266e..e33dcf82a8 100644 --- a/backend/internal/service/ops_models.go +++ b/backend/internal/service/ops_models.go @@ -1,6 +1,9 @@ package service -import "time" +import ( + "strings" + "time" +) type OpsSystemLog struct { ID int64 `json:"id"` @@ -65,17 +68,22 @@ type OpsErrorLog struct { RequestedModel string `json:"requested_model"` UpstreamModel string `json:"upstream_model"` RequestType *int16 `json:"request_type"` + UserAgent string `json:"user_agent"` // 关联 api_key 名称(LEFT JOIN api_keys 取得;软删只覆盖 key 列,name 保留,故已删 key 仍有原名)。 APIKeyName string `json:"api_key_name,omitempty"` APIKeyDeleted bool `json:"api_key_deleted,omitempty"` + + // 已删除 KEY 所有者(INVALID_API_KEY 且该 key 曾存在时的归因快照)。 + // 认证失败行 user_id 为空,列表用户列以此回退显示所有者。 + DeletedKeyOwnerUserID *int64 `json:"deleted_key_owner_user_id,omitempty"` + DeletedKeyOwnerEmail string `json:"deleted_key_owner_email,omitempty"` } type OpsErrorLogDetail struct { OpsErrorLog ErrorBody string `json:"error_body"` - UserAgent string `json:"user_agent"` // Upstream context (optional) UpstreamStatusCode *int `json:"upstream_status_code,omitempty"` @@ -93,11 +101,10 @@ type OpsErrorLogDetail struct { // vNext metric semantics IsBusinessLimited bool `json:"is_business_limited"` - // Deleted key owner info (populated when INVALID_API_KEY and key was previously deleted) - AttemptedKeyPrefix string `json:"attempted_key_prefix,omitempty"` - DeletedKeyOwnerUserID *int64 `json:"deleted_key_owner_user_id,omitempty"` - DeletedKeyOwnerEmail string `json:"deleted_key_owner_email,omitempty"` - DeletedKeyName string `json:"deleted_key_name,omitempty"` + // Deleted key owner info (populated when INVALID_API_KEY and key was previously deleted). + // OwnerUserID/OwnerEmail 已上移到 OpsErrorLog(列表用户列回退需要)。 + AttemptedKeyPrefix string `json:"attempted_key_prefix,omitempty"` + DeletedKeyName string `json:"deleted_key_name,omitempty"` // Bound (non-deleted) key prefix, snapshotted at error time; mutually exclusive with AttemptedKeyPrefix. APIKeyPrefix string `json:"api_key_prefix,omitempty"` @@ -142,8 +149,14 @@ type OpsErrorLogFilter struct { // ExcludeCountTokens drops count_tokens probe errors (is_count_tokens=true). ExcludeCountTokens bool + // IncludeRecoveredUpstream 显式豁免 status>=400 守卫(仅在 Phase=="upstream" 时生效): + // ops 专用上游错误列表需要看到 status<400 的 recovered upstream 行。 + // 请求错误语义的端点不设此开关,phase=upstream 过滤照常生效且守卫保留。 + IncludeRecoveredUpstream bool + // ErrorPhasesAny / ErrorTypesAny add plain ANY() filters WITHOUT touching the - // special-cased single `Phase` field (only Phase=="upstream" bypasses the status>=400 clause). + // special-cased single `Phase` field (only Phase=="upstream" with + // IncludeRecoveredUpstream bypasses the status>=400 clause). // NOTE: these ANY filters do NOT bypass status>=400; records with error_phase='upstream' // but status_code<400 (recovered upstream errors) remain excluded. // Used to map user-facing coarse categories to backend conditions. @@ -158,6 +171,19 @@ type OpsErrorLogFilter struct { Page int PageSize int + + // SortBy/SortOrder: server-side sorting aligned with the usage-log list. + // Repo whitelists columns (created_at/model/status_code); anything else + // falls back to created_at. SortOrder is "asc"/"desc" (default desc). + SortBy string + SortOrder string +} + +// SetSort normalizes raw sort_by/sort_order query values into the filter. +// Shared by the admin and user-facing error list handlers. +func (f *OpsErrorLogFilter) SetSort(sortBy, sortOrder string) { + f.SortBy = strings.TrimSpace(sortBy) + f.SortOrder = strings.TrimSpace(sortOrder) } type OpsErrorLogList struct { diff --git a/backend/internal/service/ops_service.go b/backend/internal/service/ops_service.go index a8c8a4bb5c..61f85ef904 100644 --- a/backend/internal/service/ops_service.go +++ b/backend/internal/service/ops_service.go @@ -359,10 +359,12 @@ func (s *OpsService) ListUserErrorRequests(ctx context.Context, userID int64, fi filter.UserQuery = "" filter.Owner = "" filter.Source = "" - // 清空 Phase 是防御:Phase 是单值特殊字段,仅当其 == "upstream" 时 buildOpsErrorLogsWhere 才跳过 status>=400 子句。 - // 用户端一律改走 category→ErrorPhasesAny/ErrorTypesAny(纯 ANY 过滤,不影响 status>=400 子句), - // 因此 recovered upstream(error_phase='upstream' 但 status<400,最终成功返回)记录对用户不可见——符合预期。 + // 清空 Phase 是防御:用户端一律改走 category→ErrorPhasesAny/ErrorTypesAny + //(纯 ANY 过滤,不影响 status>=400 子句)。守卫豁免现在还需要 + // IncludeRecoveredUpstream(用户端永不设置),recovered upstream + //(error_phase='upstream' 但 status<400,最终成功返回)记录对用户不可见——符合预期。 filter.Phase = "" + filter.IncludeRecoveredUpstream = false list, err := s.opsRepo.ListErrorLogs(ctx, filter) if err != nil { diff --git a/backend/internal/service/ops_service_user_error_test.go b/backend/internal/service/ops_service_user_error_test.go index 9027ff0788..c3b0967b67 100644 --- a/backend/internal/service/ops_service_user_error_test.go +++ b/backend/internal/service/ops_service_user_error_test.go @@ -184,16 +184,16 @@ func TestGetUserErrorRequestDetail_DeletedKeyOwnerAccess(t *testing.T) { mk := func() *OpsErrorLogDetail { return &OpsErrorLogDetail{ OpsErrorLog: OpsErrorLog{ - ID: 55, - Phase: "auth", - Type: "api_error", - StatusCode: 401, - Message: "Invalid API key", - UserID: nil, - APIKeyName: "my-old-key", - APIKeyDeleted: true, + ID: 55, + Phase: "auth", + Type: "api_error", + StatusCode: 401, + Message: "Invalid API key", + UserID: nil, + APIKeyName: "my-old-key", + APIKeyDeleted: true, + DeletedKeyOwnerUserID: &ownerUID, }, - DeletedKeyOwnerUserID: &ownerUID, } } diff --git a/backend/internal/service/ops_user_error.go b/backend/internal/service/ops_user_error.go index 7dd128afa7..e3055c2392 100644 --- a/backend/internal/service/ops_user_error.go +++ b/backend/internal/service/ops_user_error.go @@ -3,9 +3,12 @@ package service import "time" // UserErrorRequest 是面向终端用户的"错误请求"精简脱敏视图(白名单)。 -// 严禁包含 client_ip / user_agent / account / api_key_prefix / upstream_endpoint / -// user_email 等敏感或内部字段。注:message(网关标准化错误描述)与 key_name +// 严禁包含 account / api_key_prefix / upstream_endpoint / user_email 等 +// 敏感或内部字段。注:message(网关标准化错误描述)与 key_name // (用户自有 API Key 名称,KeysView 中本就可见)经产品决策对该用户开放; +// client_ip / user_agent / group_name / request_type / stream 均为该用户 +// 自己请求的属性,经产品决策(2026-07-03)开放, +// 与用量明细已向用户展示自身 ip_address/user_agent/分组/类型 的口径对齐; // error_body 仅在详情接口(GetUserErrorRequestDetail)按归属校验后返回。 type UserErrorRequest struct { ID int64 `json:"id"` @@ -18,6 +21,11 @@ type UserErrorRequest struct { Message string `json:"message"` KeyName string `json:"key_name"` KeyDeleted bool `json:"key_deleted"` + ClientIP string `json:"client_ip,omitempty"` + GroupName string `json:"group_name,omitempty"` + RequestType *int16 `json:"request_type,omitempty"` + Stream bool `json:"stream"` + UserAgent string `json:"user_agent,omitempty"` } // UserErrorRequestList 是用户错误请求分页结果。 @@ -90,6 +98,10 @@ func ToUserErrorRequest(e *OpsErrorLog) *UserErrorRequest { if model == "" { model = e.Model } + clientIP := "" + if e.ClientIP != nil { + clientIP = *e.ClientIP + } return &UserErrorRequest{ ID: e.ID, CreatedAt: e.CreatedAt, @@ -101,6 +113,11 @@ func ToUserErrorRequest(e *OpsErrorLog) *UserErrorRequest { Message: e.Message, KeyName: e.APIKeyName, KeyDeleted: e.APIKeyDeleted, + ClientIP: clientIP, + GroupName: e.GroupName, + RequestType: e.RequestType, + Stream: e.Stream, + UserAgent: e.UserAgent, } } diff --git a/backend/internal/service/ops_user_error_test.go b/backend/internal/service/ops_user_error_test.go index 31b0c26933..9e0bc164b4 100644 --- a/backend/internal/service/ops_user_error_test.go +++ b/backend/internal/service/ops_user_error_test.go @@ -122,9 +122,11 @@ func TestToUserErrorRequestDetail_WhitelistAndRedacts(t *testing.T) { UserEmail: "secret@example.com", ClientIP: func() *string { s := "1.2.3.4"; return &s }(), UpstreamEndpoint: "https://api.openai.com/v1/chat/completions", + UserAgent: "codex_cli_rs/0.125.0", + GroupName: "grp-a", + Stream: true, }, ErrorBody: `{"error":{"message":"upstream failed","type":"server_error"}}`, - UserAgent: "Mozilla/5.0 secret-agent", UpstreamStatusCode: &upstreamStatus, } @@ -147,13 +149,27 @@ func TestToUserErrorRequestDetail_WhitelistAndRedacts(t *testing.T) { t.Errorf("UpstreamStatusCode mismatch") } + // client_ip / user_agent / group_name / stream 经产品决策开放(与用量明细口径对齐) + if out.ClientIP != "1.2.3.4" { + t.Errorf("want client_ip=1.2.3.4, got %q", out.ClientIP) + } + if out.UserAgent != "codex_cli_rs/0.125.0" { + t.Errorf("want user_agent=codex_cli_rs/0.125.0, got %q", out.UserAgent) + } + if out.GroupName != "grp-a" { + t.Errorf("want group_name=grp-a, got %q", out.GroupName) + } + if !out.Stream { + t.Errorf("want stream=true") + } + // 序列化后不含敏感字段 b, err := json.Marshal(out) if err != nil { t.Fatalf("json.Marshal failed: %v", err) } raw := string(b) - for _, forbidden := range []string{"user_email", "client_ip", "upstream_endpoint", "user_agent"} { + for _, forbidden := range []string{"user_email", "upstream_endpoint"} { if strings.Contains(raw, forbidden) { t.Errorf("sensitive field %q leaked in JSON output: %s", forbidden, raw) } diff --git a/frontend/src/__tests__/setup.ts b/frontend/src/__tests__/setup.ts index b777b22e8e..9dad8c1f12 100644 --- a/frontend/src/__tests__/setup.ts +++ b/frontend/src/__tests__/setup.ts @@ -57,6 +57,20 @@ if (typeof globalThis.cancelIdleCallback === 'undefined') { }) as unknown as typeof cancelIdleCallback } +// Mock matchMedia (jsdom 未实现;DataTable 等组件依赖它做桌面/移动分支) +if (typeof window !== 'undefined' && typeof window.matchMedia !== 'function') { + window.matchMedia = ((query: string) => ({ + matches: true, // 测试默认按桌面视口渲染表格 + media: query, + onchange: null, + addListener: vi.fn(), + removeListener: vi.fn(), + addEventListener: vi.fn(), + removeEventListener: vi.fn(), + dispatchEvent: vi.fn(), + })) as unknown as typeof window.matchMedia +} + // Mock IntersectionObserver class MockIntersectionObserver { observe = vi.fn() diff --git a/frontend/src/api/admin/ops.ts b/frontend/src/api/admin/ops.ts index b3e53893ce..c7cbc64a4b 100644 --- a/frontend/src/api/admin/ops.ts +++ b/frontend/src/api/admin/ops.ts @@ -930,11 +930,16 @@ export interface OpsErrorLog { requested_model?: string upstream_model?: string request_type?: number | null + user_agent?: string + + // 已删除 KEY 所有者(INVALID_API_KEY 归因快照):认证失败行 user_id 为空, + // 用户列以此回退显示所有者 + deleted_key_owner_user_id?: number | null + deleted_key_owner_email?: string | null } export interface OpsErrorDetail extends OpsErrorLog { error_body: string - user_agent: string // Upstream context (optional; enriched by gateway services) upstream_status_code?: number | null @@ -950,10 +955,9 @@ export interface OpsErrorDetail extends OpsErrorLog { is_business_limited: boolean - // Deleted key owner info (INVALID_API_KEY attribution) + // Deleted key owner info (INVALID_API_KEY attribution); + // owner user_id/email 已上移到 OpsErrorLog(列表用户列回退) attempted_key_prefix?: string | null - deleted_key_owner_user_id?: number | null - deleted_key_owner_email?: string | null deleted_key_name?: string | null // Bound (non-deleted) key prefix, snapshotted at error time @@ -1098,6 +1102,8 @@ export type OpsErrorListQueryParams = { model?: string phase?: string + // 分类(用户侧粗分类码,如 auth/rate_limit/upstream),后端反查为 phase/type ANY 条件 + category?: string error_owner?: string error_source?: string resolved?: string @@ -1106,6 +1112,10 @@ export type OpsErrorListQueryParams = { q?: string status_codes?: string status_codes_other?: string + + // 服务端排序,列白名单见后端 opsErrorLogsOrderBy(created_at/model/status_code) + sort_by?: string + sort_order?: 'asc' | 'desc' } // Legacy unified endpoints diff --git a/frontend/src/api/admin/usage.ts b/frontend/src/api/admin/usage.ts index b37996d61a..83be033c11 100644 --- a/frontend/src/api/admin/usage.ts +++ b/frontend/src/api/admin/usage.ts @@ -86,6 +86,10 @@ export interface AdminUsageQueryParams extends UsageQueryParams { billing_mode?: string sort_by?: string sort_order?: 'asc' | 'desc' + // 错误请求 tab 专属筛选(仅传给错误列表接口;共用同一 filters 对象) + error_phase?: string | null + error_category?: string | null + status_code?: number | null } // ==================== API Functions ==================== diff --git a/frontend/src/components/admin/usage/UsageFilters.vue b/frontend/src/components/admin/usage/UsageFilters.vue index f16012ccff..bb63d9b8ee 100644 --- a/frontend/src/components/admin/usage/UsageFilters.vue +++ b/frontend/src/components/admin/usage/UsageFilters.vue @@ -121,24 +121,42 @@
- -
+ +
- -
+ +
+
+ + +
+ + +
+
@@ -156,12 +174,14 @@ {{ t('common.reset') }} - - +
@@ -172,6 +192,7 @@ import { ref, onMounted, onUnmounted, toRef, watch, computed } from 'vue' import { useI18n } from 'vue-i18n' import { adminAPI } from '@/api/admin' import Select, { type SelectOption } from '@/components/common/Select.vue' +import { COMMON_ERROR_STATUS_CODES } from '@/utils/errorBadges' import type { SimpleApiKey, SimpleUser } from '@/api/admin/usage' type ModelValue = Record @@ -183,10 +204,13 @@ interface Props { endDate: string showActions?: boolean modelOptions?: string[] + /** errors 模式:隐藏用量专属字段/按钮,显示错误类型+状态码(错误请求 tab 用) */ + mode?: 'usage' | 'errors' } const props = withDefaults(defineProps(), { - showActions: true + showActions: true, + mode: 'usage' }) const emit = defineEmits([ 'update:modelValue', @@ -243,6 +267,29 @@ const billingTypeOptions = ref([ { value: 1, label: t('admin.usage.billingTypeSubscription') } ]) +// 错误类型对应后端 phase 参数(与错误表"类型"徽章同语义) +const errorPhaseOptions = computed(() => [ + { value: null, label: t('admin.usage.allTypes') }, + { value: 'upstream', label: t('admin.ops.errorLog.typeUpstream') }, + { value: 'request', label: t('admin.ops.errorLog.typeRequest') }, + { value: 'auth', label: t('admin.ops.errorLog.typeAuth') }, + { value: 'routing', label: t('admin.ops.errorLog.typeRouting') }, + { value: 'internal', label: t('admin.ops.errorLog.typeInternal') }, +]) + +// 分类码同用户端 /usage 错误筛选;"other" 无法反查为过滤条件,刻意不列 +const errorCategoryCodes = ['auth', 'rate_limit', 'quota', 'invalid_request', 'service_unavailable', 'upstream', 'internal', 'cyber'] + +const errorCategoryOptions = computed(() => [ + { value: null, label: t('usage.errors.allCategories') }, + ...errorCategoryCodes.map((c) => ({ value: c, label: t('usage.errors.categories.' + c) })), +]) + +const statusCodeOptions = computed(() => [ + { value: null, label: t('usage.errors.allStatuses') }, + ...COMMON_ERROR_STATUS_CODES.map((c) => ({ value: c, label: String(c) })), +]) + const billingModeOptions = ref([ { value: null, label: t('admin.usage.allBillingModes') }, { value: 'token', label: t('admin.usage.billingModeToken') }, diff --git a/frontend/src/components/common/DataTable.vue b/frontend/src/components/common/DataTable.vue index bb7315dc21..d38ebeb0e1 100644 --- a/frontend/src/components/common/DataTable.vue +++ b/frontend/src/components/common/DataTable.vue @@ -36,6 +36,8 @@ v-for="(row, index) in sortedData" :key="resolveRowKey(row, index)" class="rounded-lg border border-gray-200 bg-white p-4 dark:border-dark-700 dark:bg-dark-900" + :class="{ 'cursor-pointer': clickableRows }" + @click="clickableRows && emit('rowClick', row)" >
() // 表格容器引用 @@ -381,6 +386,8 @@ interface Props { * will emit 'sort' events instead of performing client-side sorting. */ serverSideSort?: boolean + /** Emit 'rowClick' on row/card click and show pointer cursor (interactive cells should @click.stop) */ + clickableRows?: boolean /** Estimated row height in px for the virtualizer (default 56) */ estimateRowHeight?: number /** Number of rows to render beyond the visible area (default 5) */ diff --git a/frontend/src/components/common/IpGeoBatchToolbar.vue b/frontend/src/components/common/IpGeoBatchToolbar.vue new file mode 100644 index 0000000000..07199f1b2a --- /dev/null +++ b/frontend/src/components/common/IpGeoBatchToolbar.vue @@ -0,0 +1,59 @@ + + + diff --git a/frontend/src/components/user/UserErrorRequestsTable.vue b/frontend/src/components/user/UserErrorRequestsTable.vue index 42cb93ce04..2465fe2f66 100644 --- a/frontend/src/components/user/UserErrorRequestsTable.vue +++ b/frontend/src/components/user/UserErrorRequestsTable.vue @@ -1,91 +1,119 @@