Merge pull request #6602 from ZeroClover/feat/codex-manifest-pinned-accounts

feat(codex): pinned-accounts Codex model manifest for OpenAI groups
This commit is contained in:
Wesley Liddick
2026-09-05 14:51:17 +08:00
committed by GitHub
41 changed files with 2308 additions and 68 deletions
@@ -0,0 +1,15 @@
package domain
// GroupCodexModelsManifestConfig controls which accounts fetch the Codex
// models manifest for an OpenAI group. When enabled, the group's Codex
// /models requests bypass the scheduler and only use the pinned accounts.
type GroupCodexModelsManifestConfig struct {
Enabled bool `json:"enabled"`
// AccountIDs is the ordered list of pinned account IDs. Order decides
// merge precedence for duplicate model slugs.
AccountIDs []int64 `json:"account_ids,omitempty"`
// FallbackToScheduler controls the behavior when no pinned account is
// usable or all upstream fetches fail: fall back to the scheduler loop
// (true) or return the upstream error / 503 (false, default).
FallbackToScheduler bool `json:"fallback_to_scheduler,omitempty"`
}
@@ -154,6 +154,8 @@ type CreateGroupRequest struct {
DefaultMappedModel string `json:"default_mapped_model"`
MessagesDispatchModelConfig service.OpenAIMessagesDispatchModelConfig `json:"messages_dispatch_model_config"`
ModelsListConfig service.GroupModelsListConfig `json:"models_list_config"`
// 固定账号 manifest 配置;创建路径禁止开启,仅编辑可配置。
CodexModelsManifestConfig service.GroupCodexModelsManifestConfig `json:"codex_models_manifest_config"`
// 分组 RPM 上限(0 = 不限制)
RPMLimit int `json:"rpm_limit"`
// OpenAI/Codex 请求推理强度上限,空字符串表示不限制。
@@ -227,6 +229,8 @@ type UpdateGroupRequest struct {
DefaultMappedModel *string `json:"default_mapped_model"`
MessagesDispatchModelConfig *service.OpenAIMessagesDispatchModelConfig `json:"messages_dispatch_model_config"`
ModelsListConfig *service.GroupModelsListConfig `json:"models_list_config"`
// 固定账号 manifest 配置;nil 表示不修改。
CodexModelsManifestConfig *service.GroupCodexModelsManifestConfig `json:"codex_models_manifest_config"`
// 分组 RPM 上限(0 = 不限制);nil 表示未提供不改动
RPMLimit *int `json:"rpm_limit"`
// OpenAI/Codex 请求推理强度上限;空字符串清除,nil 不修改。
@@ -565,6 +569,7 @@ func (h *GroupHandler) Create(c *gin.Context) {
DefaultMappedModel: req.DefaultMappedModel,
MessagesDispatchModelConfig: req.MessagesDispatchModelConfig,
ModelsListConfig: req.ModelsListConfig,
CodexModelsManifestConfig: req.CodexModelsManifestConfig,
RPMLimit: req.RPMLimit,
MaxReasoningEffort: req.MaxReasoningEffort,
MaxReasoningEffortOverLimit: req.MaxReasoningEffortOverLimit,
@@ -697,6 +702,7 @@ func (h *GroupHandler) Update(c *gin.Context) {
DefaultMappedModel: req.DefaultMappedModel,
MessagesDispatchModelConfig: req.MessagesDispatchModelConfig,
ModelsListConfig: req.ModelsListConfig,
CodexModelsManifestConfig: req.CodexModelsManifestConfig,
RPMLimit: req.RPMLimit,
MaxReasoningEffort: req.MaxReasoningEffort,
MaxReasoningEffortOverLimit: req.MaxReasoningEffortOverLimit,
@@ -0,0 +1,41 @@
package dto
import (
"encoding/json"
"testing"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/stretchr/testify/require"
)
// 固定账号 manifest 配置必须在管理端 DTO 完整序列化:漏映射会让编辑对话框回显
// 关闭状态,管理员保存无关字段时会把已开启的配置静默关掉。用户侧 DTO 不携带该字段。
func TestGroupMapperRoundTripsCodexModelsManifestConfig(t *testing.T) {
group := &service.Group{
ID: 31, Name: "codex-manifest-dto", Platform: service.PlatformOpenAI, Status: service.StatusActive,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{101, 202},
FallbackToScheduler: true,
},
}
adminJSON, err := json.Marshal(GroupFromServiceAdmin(group))
require.NoError(t, err)
var adminEnvelope struct {
CodexModelsManifestConfig *struct {
Enabled bool `json:"enabled"`
AccountIDs []int64 `json:"account_ids"`
FallbackToScheduler bool `json:"fallback_to_scheduler"`
} `json:"codex_models_manifest_config"`
}
require.NoError(t, json.Unmarshal(adminJSON, &adminEnvelope))
require.NotNil(t, adminEnvelope.CodexModelsManifestConfig, "GroupFromServiceAdmin 漏映射 codex_models_manifest_config")
require.True(t, adminEnvelope.CodexModelsManifestConfig.Enabled)
require.Equal(t, []int64{101, 202}, adminEnvelope.CodexModelsManifestConfig.AccountIDs)
require.True(t, adminEnvelope.CodexModelsManifestConfig.FallbackToScheduler)
userJSON, err := json.Marshal(GroupFromService(group))
require.NoError(t, err)
require.NotContains(t, string(userJSON), "codex_models_manifest_config", "用户侧分组 DTO 不得暴露管理端 manifest 配置")
}
+1
View File
@@ -160,6 +160,7 @@ func GroupFromServiceAdmin(g *service.Group) *AdminGroup {
DefaultMappedModel: g.DefaultMappedModel,
MessagesDispatchModelConfig: g.MessagesDispatchModelConfig,
ModelsListConfig: g.ModelsListConfig,
CodexModelsManifestConfig: g.CodexModelsManifestConfig,
SupportedModelScopes: g.SupportedModelScopes,
AccountCount: g.AccountCount,
ActiveAccountCount: g.ActiveAccountCount,
+2
View File
@@ -190,6 +190,8 @@ type AdminGroup struct {
DefaultMappedModel string `json:"default_mapped_model"`
MessagesDispatchModelConfig domain.OpenAIMessagesDispatchModelConfig `json:"messages_dispatch_model_config"`
ModelsListConfig domain.GroupModelsListConfig `json:"models_list_config"`
// 固定账号获取 Codex Model Manifest 配置(仅 openai 平台使用)。
CodexModelsManifestConfig domain.GroupCodexModelsManifestConfig `json:"codex_models_manifest_config"`
// 支持的模型系列(仅 antigravity 平台使用)
SupportedModelScopes []string `json:"supported_model_scopes"`
@@ -1,6 +1,7 @@
package handler
import (
"errors"
"net/http"
"github.com/gin-gonic/gin"
@@ -50,6 +51,44 @@ func (h *OpenAIGatewayHandler) CodexModels(c *gin.Context) {
return
}
// 固定账号分支:开启后只用选定账号拉取 manifest,不经过调度器;
// 全部不可用/全部失败时按 FallbackToScheduler 决定回退调度器或返回错误。
if apiKey.Group.Platform == service.PlatformOpenAI &&
apiKey.Group.CodexModelsManifestConfig.Enabled &&
len(apiKey.Group.CodexModelsManifestConfig.AccountIDs) > 0 {
pinnedManifest, pinnedAccount, pinnedErr := h.gatewayService.FetchPinnedCodexModelsManifest(
c.Request.Context(),
apiKey.Group,
c.Query("client_version"),
)
if pinnedErr != nil {
if c.Request.Context().Err() != nil {
return
}
if !apiKey.Group.CodexModelsManifestConfig.FallbackToScheduler {
if errors.Is(pinnedErr, service.ErrNoPinnedCodexModelsAccounts) {
h.errorResponse(c, http.StatusServiceUnavailable, "upstream_error", "No available pinned OpenAI accounts")
return
}
h.errorResponse(c, infraerrors.Code(pinnedErr), "upstream_error", infraerrors.Message(pinnedErr))
return
}
// 回退开启:跌入下方调度器循环。
} else {
// 让 ops 错误日志携带实际拉取成功的首个固定账号。
setOpsSelectedAccount(c, pinnedAccount.ID, pinnedAccount.Platform)
if err := h.gatewayService.MergeGroupConfiguredCodexModels(c.Request.Context(), apiKey.Group, pinnedManifest, ifNoneMatch); err != nil {
h.errorResponse(c, http.StatusInternalServerError, "api_error", "Failed to build Codex models manifest")
return
}
if c.Request.Context().Err() != nil {
return
}
writeCodexModelsManifestResponse(c, pinnedManifest)
return
}
}
maxAccountSwitches := h.maxAccountSwitches
if maxAccountSwitches <= 0 {
maxAccountSwitches = 3
@@ -9,9 +9,11 @@ import (
"net"
"net/http"
"net/http/httptest"
"sort"
"strings"
"sync"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/config"
middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
@@ -640,3 +642,316 @@ func equalInt64Slices(got, want []int64) bool {
}
return true
}
// --- 固定账号 manifest 模式 ---
// codexModelsPinnedHTTPUpstream 按账号返回不同 manifest/状态的测试上游。
type codexModelsPinnedHTTPUpstream struct {
service.HTTPUpstream
mu sync.Mutex
calls []int64
bodies map[int64]string
statuses map[int64]int
}
func (u *codexModelsPinnedHTTPUpstream) Do(_ *http.Request, _ string, accountID int64, _ int) (*http.Response, error) {
u.mu.Lock()
u.calls = append(u.calls, accountID)
u.mu.Unlock()
if status, ok := u.statuses[accountID]; ok {
return &http.Response{
StatusCode: status,
Status: http.StatusText(status),
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader(`{"error":{"message":"upstream boom"}}`)),
}, nil
}
body, ok := u.bodies[accountID]
if !ok {
body = `{"models":[]}`
}
return &http.Response{
StatusCode: http.StatusOK,
Status: "200 OK",
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader(body)),
}, nil
}
func (u *codexModelsPinnedHTTPUpstream) accountIDs() []int64 {
u.mu.Lock()
defer u.mu.Unlock()
ids := append([]int64(nil), u.calls...)
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
return ids
}
func newPinnedCodexAccount(id int64, status string, schedulable bool, rateLimited bool) service.Account {
account := service.Account{
ID: id,
Name: fmt.Sprintf("pinned-%d", id),
Platform: service.PlatformOpenAI,
Type: service.AccountTypeAPIKey,
Status: status,
Schedulable: schedulable,
Priority: int(id),
Concurrency: 1,
Credentials: map[string]any{
"api_key": fmt.Sprintf("sk-pinned-%d", id),
"base_url": fmt.Sprintf("https://pinned-%d.example/v1", id),
},
}
if rateLimited {
reset := time.Now().Add(10 * time.Minute)
account.RateLimitResetAt = &reset
}
return account
}
func newPinnedCodexTestHandler(accounts []service.Account, upstream *codexModelsPinnedHTTPUpstream, maxSwitches int) *OpenAIGatewayHandler {
gin.SetMode(gin.TestMode)
cfg := &config.Config{RunMode: config.RunModeSimple}
gatewayService := service.NewOpenAIGatewayService(
codexModelsFailoverAccountRepo{accounts: accounts},
nil, nil, nil, nil, nil, nil, cfg, nil, nil, nil, nil, nil,
upstream,
nil, nil, nil, nil, nil, nil, nil, nil,
)
return &OpenAIGatewayHandler{gatewayService: gatewayService, maxAccountSwitches: maxSwitches}
}
func performPinnedCodexModelsRequest(t *testing.T, handler *OpenAIGatewayHandler, group *service.Group, etag string) *httptest.ResponseRecorder {
return performCodexModelsRequestForGroup(t, handler, group, etag)
}
func TestCodexModelsPinnedAccountsMergeUnionWithoutScheduler(t *testing.T) {
// 账号 1 优先级最高(调度器会先选它);固定配置只用 2、3,
// 返回并集且不打账号 1,即可证明没有经过调度器。
accounts := []service.Account{
newPinnedCodexAccount(1, service.StatusActive, true, false),
newPinnedCodexAccount(2, service.StatusActive, true, false),
newPinnedCodexAccount(3, service.StatusActive, true, false),
}
upstream := &codexModelsPinnedHTTPUpstream{bodies: map[int64]string{
1: `{"models":[{"slug":"scheduler-only"}]}`,
2: `{"models":[{"slug":"model-a"}]}`,
3: `{"models":[{"slug":"model-a"},{"slug":"model-c"}]}`,
}}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 77,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2, 3},
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
require.Equal(t, []string{"model-a", "model-c"}, codexHandlerManifestSlugs(t, recorder))
require.Equal(t, []int64{2, 3}, upstream.accountIDs(), "固定账号模式不得调用调度器或非固定账号")
}
func TestCodexModelsPinnedAccountsUseRateLimitedAccountAndSkipUnavailable(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(2, service.StatusActive, true, true), // 限流中:仍被使用
newPinnedCodexAccount(3, service.StatusActive, false, false), // 调度开关关闭:跳过
newPinnedCodexAccount(4, service.StatusDisabled, true, false), // 停用:跳过
newPinnedCodexAccount(5, service.StatusActive, true, false), // 正常
}
upstream := &codexModelsPinnedHTTPUpstream{bodies: map[int64]string{
2: `{"models":[{"slug":"from-rate-limited"}]}`,
5: `{"models":[{"slug":"model-five"}]}`,
}}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 78,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2, 3, 4, 5, 99}, // 99 不在分组:跳过
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
require.Equal(t, []string{"from-rate-limited", "model-five"}, codexHandlerManifestSlugs(t, recorder))
require.Equal(t, []int64{2, 5}, upstream.accountIDs())
}
func TestCodexModelsPinnedAccountsPartialFailureStillSucceeds(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(2, service.StatusActive, true, false),
newPinnedCodexAccount(3, service.StatusActive, true, false),
}
upstream := &codexModelsPinnedHTTPUpstream{
bodies: map[int64]string{2: `{"models":[{"slug":"model-a"}]}`},
statuses: map[int64]int{3: http.StatusServiceUnavailable},
}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 79,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2, 3},
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
require.Equal(t, []string{"model-a"}, codexHandlerManifestSlugs(t, recorder))
}
func TestCodexModelsPinnedAccountsAllUnavailableReturns503ByDefault(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(2, service.StatusActive, false, false),
}
upstream := &codexModelsPinnedHTTPUpstream{}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 80,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2},
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusServiceUnavailable, recorder.Code, recorder.Body.String())
require.Empty(t, upstream.accountIDs())
}
func TestCodexModelsPinnedAccountsAllFailedReturnsUpstreamError(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(2, service.StatusActive, true, false),
newPinnedCodexAccount(3, service.StatusActive, true, false),
}
upstream := &codexModelsPinnedHTTPUpstream{statuses: map[int64]int{
2: http.StatusServiceUnavailable,
3: http.StatusGatewayTimeout,
}}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 81,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2, 3},
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusBadGateway, recorder.Code, recorder.Body.String())
require.Contains(t, recorder.Body.String(), "upstream error 504", "全部失败时返回最后一个上游错误")
}
func TestCodexModelsPinnedAccountsFallbackToScheduler(t *testing.T) {
// 固定账号 2 不可用且回退开启:跌入调度器,选中优先级最高的账号 1。
accounts := []service.Account{
newPinnedCodexAccount(1, service.StatusActive, true, false),
newPinnedCodexAccount(2, service.StatusActive, false, false),
}
upstream := &codexModelsPinnedHTTPUpstream{bodies: map[int64]string{
1: `{"models":[{"slug":"from-scheduler"}]}`,
}}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 82,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2},
FallbackToScheduler: true,
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
require.Equal(t, []string{"from-scheduler"}, codexHandlerManifestSlugs(t, recorder))
require.Equal(t, []int64{1}, upstream.accountIDs())
}
func TestCodexModelsPinnedAccountsFallbackToSchedulerOnAllFailed(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(1, service.StatusActive, true, false),
newPinnedCodexAccount(2, service.StatusActive, true, false),
}
upstream := &codexModelsPinnedHTTPUpstream{
bodies: map[int64]string{1: `{"models":[{"slug":"from-scheduler"}]}`},
statuses: map[int64]int{2: http.StatusServiceUnavailable},
}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 83,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2},
FallbackToScheduler: true,
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
require.Equal(t, []string{"from-scheduler"}, codexHandlerManifestSlugs(t, recorder))
}
func TestCodexModelsPinnedAccountsStillApplyCustomModelsListFilter(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(2, service.StatusActive, true, false),
newPinnedCodexAccount(3, service.StatusActive, true, false),
}
upstream := &codexModelsPinnedHTTPUpstream{bodies: map[int64]string{
2: `{"models":[{"slug":"model-a"}]}`,
3: `{"models":[{"slug":"model-b"}]}`,
}}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 84,
Platform: service.PlatformOpenAI,
ModelsListConfig: service.GroupModelsListConfig{
Enabled: true,
Models: []string{"model-b"},
},
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2, 3},
},
}
recorder := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
require.Equal(t, []string{"model-b"}, codexHandlerManifestSlugs(t, recorder), "分组自定义模型列表过滤仍生效")
}
func TestCodexModelsPinnedAccountsETagMatchReturns304(t *testing.T) {
accounts := []service.Account{
newPinnedCodexAccount(2, service.StatusActive, true, false),
newPinnedCodexAccount(3, service.StatusActive, true, false),
}
upstream := &codexModelsPinnedHTTPUpstream{bodies: map[int64]string{
2: `{"models":[{"slug":"model-a"}]}`,
3: `{"models":[{"slug":"model-c"}]}`,
}}
handler := newPinnedCodexTestHandler(accounts, upstream, 3)
group := &service.Group{
ID: 85,
Platform: service.PlatformOpenAI,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{2, 3},
},
}
first := performPinnedCodexModelsRequest(t, handler, group, "")
require.Equal(t, http.StatusOK, first.Code, first.Body.String())
etag := first.Header().Get("ETag")
require.NotEmpty(t, etag)
second := performPinnedCodexModelsRequest(t, handler, group, etag)
require.Equal(t, http.StatusNotModified, second.Code, second.Body.String())
require.Empty(t, second.Body.Bytes())
}
@@ -215,6 +215,7 @@ func (r *apiKeyRepository) GetByKeyForAuth(ctx context.Context, key string) (*se
group.FieldDefaultMappedModel,
group.FieldMessagesDispatchModelConfig,
group.FieldModelsListConfig,
group.FieldCodexModelsManifestConfig,
group.FieldRpmLimit,
group.FieldMaxReasoningEffort,
group.FieldMaxReasoningEffortOverLimit,
@@ -1017,6 +1018,7 @@ func groupEntityToService(g *dbent.Group) *service.Group {
DefaultMappedModel: g.DefaultMappedModel,
MessagesDispatchModelConfig: g.MessagesDispatchModelConfig,
ModelsListConfig: g.ModelsListConfig,
CodexModelsManifestConfig: g.CodexModelsManifestConfig,
RPMLimit: g.RpmLimit,
MaxReasoningEffort: g.MaxReasoningEffort,
MaxReasoningEffortOverLimit: g.MaxReasoningEffortOverLimit,
@@ -0,0 +1,52 @@
//go:build integration
package repository
import (
"context"
"fmt"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/stretchr/testify/require"
)
func TestGetByKeyForAuthCarriesGroupCodexModelsManifestConfig(t *testing.T) {
ctx := context.Background()
suffix := time.Now().UnixNano()
group := mustCreateGroup(t, integrationEntClient, &service.Group{
Name: fmt.Sprintf("codex-manifest-proj-group-%d", suffix), Platform: service.PlatformOpenAI,
RateMultiplier: 1,
CodexModelsManifestConfig: service.GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{11, 22},
FallbackToScheduler: true,
},
})
user := mustCreateUser(t, integrationEntClient, &service.User{
Email: fmt.Sprintf("codex-manifest-proj-%d@example.com", suffix), Concurrency: 5,
})
groupID := group.ID
keyValue := fmt.Sprintf("sk-codex-manifest-proj-%d", suffix)
apiKeyRepo := NewAPIKeyRepository(integrationEntClient, integrationDB)
key := &service.APIKey{UserID: user.ID, GroupID: &groupID, Key: keyValue, Name: "codex-manifest-proj", Status: service.StatusActive}
require.NoError(t, apiKeyRepo.Create(ctx, key))
t.Cleanup(func() {
_, err := integrationDB.ExecContext(ctx, "DELETE FROM auth_cache_invalidation_outbox WHERE cache_key = encode(sha256(convert_to($1, 'UTF8')), 'hex')", keyValue)
require.NoError(t, err)
_, err = integrationDB.ExecContext(ctx, "DELETE FROM api_keys WHERE id = $1", key.ID)
require.NoError(t, err)
_, err = integrationDB.ExecContext(ctx, "DELETE FROM users WHERE id = $1", user.ID)
require.NoError(t, err)
_, err = integrationDB.ExecContext(ctx, "DELETE FROM groups WHERE id = $1", group.ID)
require.NoError(t, err)
})
got, err := apiKeyRepo.GetByKeyForAuth(ctx, keyValue)
require.NoError(t, err)
require.NotNil(t, got.Group)
require.True(t, got.Group.CodexModelsManifestConfig.Enabled, "codex_models_manifest_config 必须进入认证投影(投影漏列会让固定账号模式静默失效)")
require.Equal(t, []int64{11, 22}, got.Group.CodexModelsManifestConfig.AccountIDs)
require.True(t, got.Group.CodexModelsManifestConfig.FallbackToScheduler)
}
@@ -92,6 +92,7 @@ func mustCreateGroup(t *testing.T, client *dbent.Client, g *service.Group) *serv
SetIsExclusive(g.IsExclusive).
SetForceOpenaiFast(g.ForceOpenAIFast).
SetFreeOpenaiFast(g.FreeOpenAIFast).
SetCodexModelsManifestConfig(g.CodexModelsManifestConfig).
SetProfitControlEnabled(g.ProfitControlEnabled).
SetProfitMinMargin(g.ProfitMinMargin).
SetProfitSafetyBuffer(g.ProfitSafetyBuffer)
@@ -110,6 +110,7 @@ func createGroupRecord(ctx context.Context, client *dbent.Client, groupIn *servi
SetDefaultMappedModel(groupIn.DefaultMappedModel).
SetMessagesDispatchModelConfig(groupIn.MessagesDispatchModelConfig).
SetModelsListConfig(groupIn.ModelsListConfig).
SetCodexModelsManifestConfig(groupIn.CodexModelsManifestConfig).
SetRpmLimit(groupIn.RPMLimit).
SetMaxReasoningEffort(groupIn.MaxReasoningEffort).
SetMaxReasoningEffortOverLimit(groupIn.MaxReasoningEffortOverLimit).
@@ -289,6 +290,7 @@ func (r *groupRepository) Update(ctx context.Context, groupIn *service.Group) er
SetDefaultMappedModel(groupIn.DefaultMappedModel).
SetMessagesDispatchModelConfig(groupIn.MessagesDispatchModelConfig).
SetModelsListConfig(groupIn.ModelsListConfig).
SetCodexModelsManifestConfig(groupIn.CodexModelsManifestConfig).
SetRpmLimit(groupIn.RPMLimit).
SetMaxReasoningEffort(groupIn.MaxReasoningEffort).
SetMaxReasoningEffortOverLimit(groupIn.MaxReasoningEffortOverLimit).
+24 -4
View File
@@ -314,6 +314,11 @@ func (s *adminServiceImpl) CreateGroup(ctx context.Context, input *CreateGroupIn
}
platform := NormalizeGroupPlatform(input.Platform)
// 固定账号 manifest 配置:账号绑定发生在创建之后,创建时无法校验成员关系,
// 拒绝开启并在创建后的编辑里配置。
if normalizeCodexModelsManifestConfig(platform, input.CodexModelsManifestConfig).Enabled {
return nil, infraerrors.New(http.StatusBadRequest, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", "codex models manifest config cannot be enabled at group creation; configure it after creation in the group editor")
}
modelPricing, err := normalizeGroupModelPricing(platform, input.ModelPricing)
if err != nil {
return nil, err
@@ -524,10 +529,13 @@ func (s *adminServiceImpl) CreateGroup(ctx context.Context, input *CreateGroupIn
DefaultMappedModel: input.DefaultMappedModel,
MessagesDispatchModelConfig: normalizeOpenAIMessagesDispatchModelConfig(input.MessagesDispatchModelConfig),
ModelsListConfig: normalizeGroupModelsListConfig(input.ModelsListConfig),
RPMLimit: input.RPMLimit,
MaxReasoningEffort: maxReasoningEffort,
MaxReasoningEffortOverLimit: maxReasoningEffortOverLimit,
ReasoningEffortMappings: reasoningEffortMappings,
// 固定账号 manifest 配置:账号绑定发生在分组创建之后,创建路径禁止开启,
// 成员关系无从校验(前端创建对话框也不展示)。
CodexModelsManifestConfig: normalizeCodexModelsManifestConfig(platform, input.CodexModelsManifestConfig),
RPMLimit: input.RPMLimit,
MaxReasoningEffort: maxReasoningEffort,
MaxReasoningEffortOverLimit: maxReasoningEffortOverLimit,
ReasoningEffortMappings: reasoningEffortMappings,
}
sanitizeGroupMessagesDispatchFields(group)
sanitizeGroupOpenAIFast(group)
@@ -904,6 +912,9 @@ func (s *adminServiceImpl) UpdateGroup(ctx context.Context, id int64, input *Upd
if input.ModelsListConfig != nil {
group.ModelsListConfig = normalizeGroupModelsListConfig(*input.ModelsListConfig)
}
if input.CodexModelsManifestConfig != nil {
group.CodexModelsManifestConfig = *input.CodexModelsManifestConfig
}
if input.RPMLimit != nil {
group.RPMLimit = *input.RPMLimit
}
@@ -934,6 +945,15 @@ func (s *adminServiceImpl) UpdateGroup(ctx context.Context, id int64, input *Upd
group.AllowLive = false
}
sanitizeGroupReasoningEffortPolicy(group)
// 固定账号 manifest 配置:按最终平台归一化(切出 openai 平台时静默归零,
// 与 ForceOpenAIFast 同一收口);校验仅在本次显式携带配置时进行,
// 避免脏 ID 阻塞无关字段更新。
group.CodexModelsManifestConfig = normalizeCodexModelsManifestConfig(group.Platform, group.CodexModelsManifestConfig)
if input.CodexModelsManifestConfig != nil {
if err := s.validateCodexModelsManifestConfig(ctx, id, group.CodexModelsManifestConfig); err != nil {
return nil, err
}
}
if err := s.groupRepo.Update(ctx, group); err != nil {
return nil, err
@@ -154,6 +154,8 @@ func cloneGroupForDuplicate(source *Group, operationID string) *Group {
Enabled: source.ModelsListConfig.Enabled,
Models: append([]string(nil), source.ModelsListConfig.Models...),
},
// 固定账号 manifest 配置指向源分组的账号 ID,复制后成员关系可能变化,重置为关闭且列表为空。
CodexModelsManifestConfig: GroupCodexModelsManifestConfig{},
RPMLimit: source.RPMLimit,
MaxReasoningEffort: source.MaxReasoningEffort,
MaxReasoningEffortOverLimit: source.MaxReasoningEffortOverLimit,
@@ -273,6 +273,8 @@ type CreateGroupInput struct {
RequirePrivacySet bool
MessagesDispatchModelConfig OpenAIMessagesDispatchModelConfig
ModelsListConfig GroupModelsListConfig
// CodexModelsManifestConfig 固定账号 manifest 配置;创建路径禁止开启,仅编辑可配置。
CodexModelsManifestConfig GroupCodexModelsManifestConfig
// RPMLimit 分组 RPM 上限(0 = 不限制)
RPMLimit int
// MaxReasoningEffort OpenAI/Codex 请求的推理强度上限,空字符串表示不限制。
@@ -352,6 +354,8 @@ type UpdateGroupInput struct {
RequirePrivacySet *bool
MessagesDispatchModelConfig *OpenAIMessagesDispatchModelConfig
ModelsListConfig *GroupModelsListConfig
// CodexModelsManifestConfig nil 表示不修改;非 openai 平台会被归一化为关闭。
CodexModelsManifestConfig *GroupCodexModelsManifestConfig
// RPMLimit 分组 RPM 上限(0 = 不限制),nil 表示未提供不改动。
RPMLimit *int
// MaxReasoningEffort 空字符串表示清除上限;nil 表示未提供不改动。
@@ -1901,3 +1901,208 @@ func TestAdminService_PreviewCompositeRouteUsesExplicitRoutes(t *testing.T) {
require.NotNil(t, decision.Route)
require.Equal(t, int64(11), decision.Route.ID)
}
// accountRepoStubForGroupCodexManifest 支撑固定账号 manifest 配置校验测试。
type accountRepoStubForGroupCodexManifest struct {
AccountRepository
accounts []Account
}
func (s *accountRepoStubForGroupCodexManifest) ListByGroup(_ context.Context, _ int64) ([]Account, error) {
return append([]Account(nil), s.accounts...), nil
}
func openAIGroupCodexManifestAccounts(ids ...int64) []Account {
accounts := make([]Account, 0, len(ids))
for _, id := range ids {
accounts = append(accounts, Account{
ID: id, Platform: PlatformOpenAI, Type: AccountTypeOAuth,
Status: StatusActive, Schedulable: true,
})
}
return accounts
}
func TestAdminService_UpdateGroup_CodexModelsManifestConfigValidation(t *testing.T) {
t.Run("enabled with empty list rejected", func(t *testing.T) {
existing := &Group{ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive}
repo := &groupRepoStubForAdmin{getByID: existing}
svc := &adminServiceImpl{
groupRepo: repo,
accountRepo: &accountRepoStubForGroupCodexManifest{},
}
_, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{
CodexModelsManifestConfig: &GroupCodexModelsManifestConfig{Enabled: true},
})
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, infraerrors.Code(err))
require.Equal(t, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", infraerrors.Reason(err))
require.Nil(t, repo.updated)
})
t.Run("non-member or non-openai account rejected", func(t *testing.T) {
existing := &Group{ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive}
repo := &groupRepoStubForAdmin{getByID: existing}
anthropicMember := Account{ID: 30, Platform: PlatformAnthropic, Type: AccountTypeOAuth, Status: StatusActive}
inactiveMember := Account{ID: 40, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusDisabled}
svc := &adminServiceImpl{
groupRepo: repo,
accountRepo: &accountRepoStubForGroupCodexManifest{accounts: append(openAIGroupCodexManifestAccounts(10, 20), anthropicMember, inactiveMember)},
}
_, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{
CodexModelsManifestConfig: &GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{10, 30, 40, 99},
},
})
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, infraerrors.Code(err))
require.Equal(t, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", infraerrors.Reason(err))
require.Contains(t, err.Error(), "30")
require.Contains(t, err.Error(), "40")
require.Contains(t, err.Error(), "99")
require.Nil(t, repo.updated)
})
t.Run("more than ten accounts rejected", func(t *testing.T) {
existing := &Group{ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive}
repo := &groupRepoStubForAdmin{getByID: existing}
ids := make([]int64, 0, 11)
for id := int64(1); id <= 11; id++ {
ids = append(ids, id)
}
svc := &adminServiceImpl{
groupRepo: repo,
accountRepo: &accountRepoStubForGroupCodexManifest{accounts: openAIGroupCodexManifestAccounts(ids...)},
}
_, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{
CodexModelsManifestConfig: &GroupCodexModelsManifestConfig{Enabled: true, AccountIDs: ids},
})
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, infraerrors.Code(err))
require.Equal(t, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", infraerrors.Reason(err))
require.Nil(t, repo.updated)
})
t.Run("duplicate ids deduplicated preserving order", func(t *testing.T) {
existing := &Group{ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive}
repo := &groupRepoStubForAdmin{getByID: existing}
svc := &adminServiceImpl{
groupRepo: repo,
accountRepo: &accountRepoStubForGroupCodexManifest{accounts: openAIGroupCodexManifestAccounts(1, 2, 3)},
}
group, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{
CodexModelsManifestConfig: &GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{3, 1, 3, 2, 1},
},
})
require.NoError(t, err)
require.Equal(t, []int64{3, 1, 2}, repo.updated.CodexModelsManifestConfig.AccountIDs)
require.True(t, group.CodexModelsManifestConfig.Enabled)
})
t.Run("valid config accepted", func(t *testing.T) {
existing := &Group{ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive}
repo := &groupRepoStubForAdmin{getByID: existing}
svc := &adminServiceImpl{
groupRepo: repo,
accountRepo: &accountRepoStubForGroupCodexManifest{accounts: openAIGroupCodexManifestAccounts(10, 20)},
}
group, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{
CodexModelsManifestConfig: &GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{10, 20},
FallbackToScheduler: true,
},
})
require.NoError(t, err)
require.True(t, group.CodexModelsManifestConfig.Enabled)
require.Equal(t, []int64{10, 20}, repo.updated.CodexModelsManifestConfig.AccountIDs)
require.True(t, repo.updated.CodexModelsManifestConfig.FallbackToScheduler)
})
}
func TestAdminService_UpdateGroup_CodexModelsManifestConfigZeroedForNonOpenAIPlatform(t *testing.T) {
existing := &Group{ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive}
repo := &groupRepoStubForAdmin{getByID: existing}
svc := &adminServiceImpl{
groupRepo: repo,
accountRepo: &accountRepoStubForGroupCodexManifest{},
}
group, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{
Platform: PlatformAnthropic,
CodexModelsManifestConfig: &GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{1, 2},
FallbackToScheduler: true,
},
})
require.NoError(t, err)
require.Equal(t, GroupCodexModelsManifestConfig{}, repo.updated.CodexModelsManifestConfig)
require.Equal(t, GroupCodexModelsManifestConfig{}, group.CodexModelsManifestConfig)
}
func TestAdminService_UpdateGroup_CodexModelsManifestConfigUntouchedWhenOmitted(t *testing.T) {
existing := &Group{
ID: 1, Name: "g", Platform: PlatformOpenAI, Status: StatusActive,
CodexModelsManifestConfig: GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{10},
},
}
repo := &groupRepoStubForAdmin{getByID: existing}
svc := &adminServiceImpl{groupRepo: repo}
_, err := svc.UpdateGroup(context.Background(), existing.ID, &UpdateGroupInput{Name: "renamed"})
require.NoError(t, err)
require.Equal(t, existing.CodexModelsManifestConfig, repo.updated.CodexModelsManifestConfig)
}
func TestAdminService_CreateGroup_CodexModelsManifestConfigEnabledRejected(t *testing.T) {
repo := &groupRepoStubForAdmin{createID: 61}
svc := &adminServiceImpl{groupRepo: repo}
_, err := svc.CreateGroup(context.Background(), &CreateGroupInput{
Name: "codex-pinned", Platform: PlatformOpenAI, RateMultiplier: 1,
CodexModelsManifestConfig: GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{1},
},
})
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, infraerrors.Code(err))
require.Equal(t, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", infraerrors.Reason(err))
require.Nil(t, repo.created)
}
func TestAdminService_CreateGroup_CodexModelsManifestConfigDisabledAccepted(t *testing.T) {
repo := &groupRepoStubForAdmin{createID: 62}
svc := &adminServiceImpl{groupRepo: repo}
group, err := svc.CreateGroup(context.Background(), &CreateGroupInput{
Name: "codex-pinned-off", Platform: PlatformOpenAI, RateMultiplier: 1,
CodexModelsManifestConfig: GroupCodexModelsManifestConfig{
Enabled: false,
AccountIDs: []int64{1, 1, 2},
},
})
require.NoError(t, err)
require.Equal(t, []int64{1, 2}, group.CodexModelsManifestConfig.AccountIDs)
require.False(t, repo.created.CodexModelsManifestConfig.Enabled)
}
@@ -108,6 +108,9 @@ type APIKeyAuthGroupSnapshot struct {
DefaultMappedModel string `json:"default_mapped_model,omitempty"`
MessagesDispatchModelConfig OpenAIMessagesDispatchModelConfig `json:"messages_dispatch_model_config,omitempty"`
ModelsListConfig GroupModelsListConfig `json:"models_list_config,omitempty"`
// CodexModelsManifestConfig 与 ModelsListConfig 一样在认证快照分组里透传,
// Codex /models handler 直接读认证分组对象。
CodexModelsManifestConfig GroupCodexModelsManifestConfig `json:"codex_models_manifest_config,omitempty"`
// RPMLimit 分组级每分钟请求数上限(0 = 不限制);用于 billing_cache_service.checkRPM 级联判断。
RPMLimit int `json:"rpm_limit"`
@@ -0,0 +1,43 @@
package service
import (
"context"
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
)
// 固定账号 manifest 配置必须在认证快照 JSON 往返后完整保留,
// 否则开关在缓存路径上会静默失效(投影对账由集成测试兜底,此处覆盖快照序列化)。
func TestAPIKeyAuthSnapshotGroupCodexModelsManifestRoundtrip(t *testing.T) {
groupID := int64(60)
apiKey := &APIKey{
ID: 92, UserID: 46, GroupID: &groupID, Key: "sk-codex-manifest-roundtrip", Status: StatusActive,
User: &User{ID: 46, Status: StatusActive},
Group: &Group{
ID: groupID, Name: "codex-manifest-roundtrip", Platform: PlatformOpenAI, Status: StatusActive,
Hydrated: true,
CodexModelsManifestConfig: GroupCodexModelsManifestConfig{
Enabled: true,
AccountIDs: []int64{7, 8},
FallbackToScheduler: true,
},
},
}
svc := &APIKeyService{}
payload, err := json.Marshal(&APIKeyAuthCacheEntry{Snapshot: svc.snapshotFromAPIKey(context.Background(), apiKey)})
require.NoError(t, err)
var cached APIKeyAuthCacheEntry
require.NoError(t, json.Unmarshal(payload, &cached))
materialized, used, err := svc.applyAuthCacheEntry(apiKey.Key, &cached)
require.NoError(t, err)
require.True(t, used)
require.NotNil(t, materialized.Group)
require.True(t, materialized.Group.CodexModelsManifestConfig.Enabled)
require.Equal(t, []int64{7, 8}, materialized.Group.CodexModelsManifestConfig.AccountIDs)
require.True(t, materialized.Group.CodexModelsManifestConfig.FallbackToScheduler)
require.Equal(t, apiKeyAuthSnapshotVersion, cached.Snapshot.Version)
}
@@ -14,7 +14,7 @@ import (
"github.com/dgraph-io/ristretto"
)
const apiKeyAuthSnapshotVersion = 22 // v22: group free_openai_fast field
const apiKeyAuthSnapshotVersion = 23 // v23: group codex_models_manifest_config field
type apiKeyAuthCacheConfig struct {
l1Size int
@@ -423,6 +423,7 @@ func (s *APIKeyService) snapshotFromAPIKey(ctx context.Context, apiKey *APIKey)
DefaultMappedModel: apiKey.Group.DefaultMappedModel,
MessagesDispatchModelConfig: apiKey.Group.MessagesDispatchModelConfig,
ModelsListConfig: apiKey.Group.ModelsListConfig,
CodexModelsManifestConfig: apiKey.Group.CodexModelsManifestConfig,
RPMLimit: apiKey.Group.RPMLimit,
MaxReasoningEffort: apiKey.Group.MaxReasoningEffort,
MaxReasoningEffortOverLimit: apiKey.Group.MaxReasoningEffortOverLimit,
@@ -524,6 +525,7 @@ func (s *APIKeyService) snapshotToAPIKey(key string, snapshot *APIKeyAuthSnapsho
DefaultMappedModel: snapshot.Group.DefaultMappedModel,
MessagesDispatchModelConfig: snapshot.Group.MessagesDispatchModelConfig,
ModelsListConfig: snapshot.Group.ModelsListConfig,
CodexModelsManifestConfig: snapshot.Group.CodexModelsManifestConfig,
RPMLimit: snapshot.Group.RPMLimit,
MaxReasoningEffort: snapshot.Group.MaxReasoningEffort,
MaxReasoningEffortOverLimit: snapshot.Group.MaxReasoningEffortOverLimit,
@@ -32,5 +32,5 @@ func TestAPIKeyAuthSnapshotGroupForceOpenAIFastRoundtrip(t *testing.T) {
require.True(t, materialized.Group.Hydrated)
require.True(t, materialized.Group.ForceOpenAIFast)
require.True(t, materialized.Group.FreeOpenAIFast)
require.Equal(t, 22, cached.Snapshot.Version)
require.Equal(t, apiKeyAuthSnapshotVersion, cached.Snapshot.Version)
}
@@ -53,7 +53,7 @@ func TestAPIKeyAuthSnapshotProfitControlRoundtrip(t *testing.T) {
snapshot := svc.snapshotFromAPIKey(context.Background(), apiKey)
require.NotNil(t, snapshot)
require.Equal(t, apiKeyAuthSnapshotVersion, snapshot.Version)
require.Equal(t, 22, snapshot.Version, "v22 起认证快照携带分组免费 Fast 开关")
require.Equal(t, apiKeyAuthSnapshotVersion, snapshot.Version, "认证快照版本必须与当前常量一致")
// 模拟 L2 缓存的完整 JSON 往返(与 apiKeyCache.SetAuthCache/GetAuthCache 同构)。
payload, err := json.Marshal(&APIKeyAuthCacheEntry{Snapshot: snapshot})
+4
View File
@@ -13,6 +13,7 @@ import (
type OpenAIMessagesDispatchModelConfig = domain.OpenAIMessagesDispatchModelConfig
type GroupModelsListConfig = domain.GroupModelsListConfig
type GroupCodexModelsManifestConfig = domain.GroupCodexModelsManifestConfig
type ReasoningEffortMapping = domain.ReasoningEffortMapping
type Group struct {
@@ -107,6 +108,9 @@ type Group struct {
DefaultMappedModel string
MessagesDispatchModelConfig OpenAIMessagesDispatchModelConfig
ModelsListConfig GroupModelsListConfig
// CodexModelsManifestConfig 开启后,该分组的 Codex /models manifest 请求只用
// 固定账号列表拉取并合并,不经过调度器(仅 openai 平台)。
CodexModelsManifestConfig GroupCodexModelsManifestConfig
// RPMLimit 分组级每分钟请求数上限(0 = 不限制)。
// 一旦设置即接管该分组用户的限流(覆盖用户级 rpm_limit),可被 user-group rpm_override 进一步覆盖。
@@ -0,0 +1,86 @@
package service
import (
"context"
"fmt"
"net/http"
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
)
// maxCodexModelsManifestAccounts is the upper bound of pinned accounts per group.
const maxCodexModelsManifestAccounts = 10
// normalizeCodexModelsManifestConfig normalizes a group's pinned-accounts Codex
// manifest config before persistence:
// - Non-OpenAI platforms always persist a disabled zero config (spec: silently
// normalize instead of rejecting).
// - Account IDs are de-duplicated preserving first-seen order; invalid IDs (<= 0)
// are dropped. The list is kept even when disabled so toggling enabled back
// on does not lose the previous selection.
func normalizeCodexModelsManifestConfig(platform string, cfg GroupCodexModelsManifestConfig) GroupCodexModelsManifestConfig {
if platform != PlatformOpenAI {
return GroupCodexModelsManifestConfig{}
}
out := GroupCodexModelsManifestConfig{
Enabled: cfg.Enabled,
FallbackToScheduler: cfg.FallbackToScheduler,
}
if len(cfg.AccountIDs) == 0 {
return out
}
seen := make(map[int64]struct{}, len(cfg.AccountIDs))
out.AccountIDs = make([]int64, 0, len(cfg.AccountIDs))
for _, id := range cfg.AccountIDs {
if id <= 0 {
continue
}
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
out.AccountIDs = append(out.AccountIDs, id)
}
if len(out.AccountIDs) == 0 {
out.AccountIDs = nil
}
return out
}
// validateCodexModelsManifestConfig enforces the admin contract for an enabled
// pinned-accounts config: at least one account, at most
// maxCodexModelsManifestAccounts, and every account must be an active OpenAI
// member of the group. Validation failures return 400 with reason
// INVALID_CODEX_MODELS_MANIFEST_CONFIG and never reach persistence.
func (s *adminServiceImpl) validateCodexModelsManifestConfig(ctx context.Context, groupID int64, cfg GroupCodexModelsManifestConfig) error {
if !cfg.Enabled {
return nil
}
if len(cfg.AccountIDs) == 0 {
return infraerrors.New(http.StatusBadRequest, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", "codex models manifest config requires at least one account")
}
if len(cfg.AccountIDs) > maxCodexModelsManifestAccounts {
return infraerrors.Newf(http.StatusBadRequest, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", "codex models manifest config allows at most %d accounts, got %d", maxCodexModelsManifestAccounts, len(cfg.AccountIDs))
}
accounts, err := s.accountRepo.ListByGroup(ctx, groupID)
if err != nil {
return fmt.Errorf("load group accounts for codex models manifest config: %w", err)
}
members := make(map[int64]struct{}, len(accounts))
for i := range accounts {
acc := &accounts[i]
if acc.IsActive() && acc.Platform == PlatformOpenAI {
members[acc.ID] = struct{}{}
}
}
var invalid []int64
for _, id := range cfg.AccountIDs {
if _, ok := members[id]; !ok {
invalid = append(invalid, id)
}
}
if len(invalid) > 0 {
return infraerrors.Newf(http.StatusBadRequest, "INVALID_CODEX_MODELS_MANIFEST_CONFIG", "codex models manifest config contains accounts not in this group or not active openai accounts: %v", invalid)
}
return nil
}
@@ -0,0 +1,214 @@
package service
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"strings"
"sync"
"time"
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
)
// ErrNoPinnedCodexModelsAccounts signals that an enabled pinned-accounts
// manifest config has no usable account right now (inactive, unschedulable,
// or expired members are skipped, and remaining IDs are no longer bound).
var ErrNoPinnedCodexModelsAccounts = errors.New("no usable pinned codex models manifest accounts")
// isPinnedCodexModelsAccountUsable decides whether a pinned account may fetch
// the manifest. Deliberately narrower than Account.IsSchedulable(): priority,
// load factor, rate-limit windows, overload windows and temporary
// unschedulable cooldowns are ignored, because the pinned mode exists to keep
// the manifest deterministic regardless of scheduler state.
func isPinnedCodexModelsAccountUsable(account *Account) bool {
if account == nil || !account.IsActive() || !account.Schedulable {
return false
}
if account.AutoPauseOnExpired && account.ExpiresAt != nil && !account.ExpiresAt.After(time.Now()) {
return false
}
return true
}
// mergeCodexModelsManifestBodies merges manifest bodies from the pinned
// accounts: the first body's top-level envelope is the base, and the models
// arrays are unioned by slug with first-seen (config-order) entries winning.
// Entries without a usable slug are kept once in order of appearance.
func mergeCodexModelsManifestBodies(bodies [][]byte) ([]byte, error) {
if len(bodies) == 0 {
return nil, errors.New("no codex models manifest bodies to merge")
}
var base map[string]json.RawMessage
if err := json.Unmarshal(bodies[0], &base); err != nil {
return nil, fmt.Errorf("decode base codex models manifest: %w", err)
}
if base == nil {
return nil, errors.New("base codex models manifest is not a JSON object")
}
mergedModels := make([]json.RawMessage, 0, len(bodies)*8)
seenSlug := make(map[string]struct{})
seenSlugLess := make(map[string]struct{})
for _, body := range bodies {
var envelope map[string]json.RawMessage
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, fmt.Errorf("decode codex models manifest body: %w", err)
}
modelsRaw, ok := envelope["models"]
if !ok {
continue
}
var models []json.RawMessage
if err := json.Unmarshal(modelsRaw, &models); err != nil {
return nil, fmt.Errorf("decode codex models manifest models array: %w", err)
}
for _, raw := range models {
var model struct {
Slug string `json:"slug"`
}
slug := ""
if err := json.Unmarshal(raw, &model); err == nil {
slug = strings.TrimSpace(model.Slug)
}
if slug != "" {
if _, exists := seenSlug[slug]; exists {
continue
}
seenSlug[slug] = struct{}{}
} else {
fingerprint := string(bytes.TrimSpace(raw))
if _, exists := seenSlugLess[fingerprint]; exists {
continue
}
seenSlugLess[fingerprint] = struct{}{}
}
mergedModels = append(mergedModels, raw)
}
}
encodedModels, err := json.Marshal(mergedModels)
if err != nil {
return nil, fmt.Errorf("encode merged codex models: %w", err)
}
base["models"] = encodedModels
merged, err := json.Marshal(base)
if err != nil {
return nil, fmt.Errorf("encode merged codex models manifest: %w", err)
}
return merged, nil
}
// FetchPinnedCodexModelsManifest fetches the Codex models manifest using only
// the accounts pinned on the group's codex_models_manifest_config, ignoring
// scheduler priorities, load factors and rate-limit/overload windows. All
// usable accounts are fetched concurrently; successful bodies are merged by
// slug with config-order precedence. It returns the merged manifest and the
// first successful account (in config order) for ops attribution.
//
// - no usable account → ErrNoPinnedCodexModelsAccounts
// - every fetch failed → the last upstream error
// - partial failure → successful accounts are still merged and a warning
// naming the failed account IDs is logged.
func (s *OpenAIGatewayService) FetchPinnedCodexModelsManifest(ctx context.Context, group *Group, clientVersion string) (*CodexModelsManifest, *Account, error) {
if s == nil || s.accountRepo == nil || group == nil {
return nil, nil, ErrNoPinnedCodexModelsAccounts
}
cfg := group.CodexModelsManifestConfig
if group.Platform != PlatformOpenAI || !cfg.Enabled || len(cfg.AccountIDs) == 0 {
return nil, nil, ErrNoPinnedCodexModelsAccounts
}
members, err := s.accountRepo.ListByGroup(ctx, group.ID)
if err != nil {
return nil, nil, fmt.Errorf("load pinned codex models manifest accounts: %w", err)
}
memberByID := make(map[int64]Account, len(members))
for _, member := range members {
memberByID[member.ID] = member
}
// 按配置顺序筛选可用账号;已解绑/已删除的 ID 直接跳过。
usable := make([]Account, 0, len(cfg.AccountIDs))
for _, id := range cfg.AccountIDs {
member, ok := memberByID[id]
if !ok || member.Platform != PlatformOpenAI {
continue
}
if !isPinnedCodexModelsAccountUsable(&member) {
continue
}
usable = append(usable, member)
}
if len(usable) == 0 {
return nil, nil, ErrNoPinnedCodexModelsAccounts
}
bodies := make([][]byte, len(usable))
fetchErrs := make([]error, len(usable))
// 各自独立完成、不取消兄弟请求,错误按下标收集,普通 WaitGroup 足够。
var fetchGroup sync.WaitGroup
for i := range usable {
account := usable[i]
index := i
fetchGroup.Add(1)
go func() {
defer fetchGroup.Done()
manifest, fetchErr := s.FetchCodexModelsManifest(ctx, &account, clientVersion, "")
if fetchErr != nil {
fetchErrs[index] = fetchErr
return
}
if completeErr := s.CompleteAPIKeyCodexModelsManifestForClient(manifest, &account); completeErr != nil {
fetchErrs[index] = completeErr
return
}
bodies[index] = manifest.Body
}()
}
fetchGroup.Wait()
successBodies := make([][]byte, 0, len(usable))
successAccounts := make([]*Account, 0, len(usable))
failedIDs := make([]int64, 0)
for i := range usable {
if fetchErrs[i] != nil || bodies[i] == nil {
failedIDs = append(failedIDs, usable[i].ID)
continue
}
successBodies = append(successBodies, bodies[i])
successAccounts = append(successAccounts, &usable[i])
}
if len(successBodies) == 0 {
var lastErr error
for i := range fetchErrs {
if fetchErrs[i] != nil {
lastErr = fetchErrs[i]
}
}
if lastErr == nil {
lastErr = infraerrors.New(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "pinned codex models manifest accounts all failed")
}
return nil, nil, lastErr
}
if len(failedIDs) > 0 {
slog.Warn("codex_models_manifest_pinned_partial_failure",
"group_id", group.ID,
"failed_account_ids", failedIDs,
)
}
merged, err := mergeCodexModelsManifestBodies(successBodies)
if err != nil {
return nil, nil, fmt.Errorf("merge pinned codex models manifests: %w", err)
}
firstAccount := *successAccounts[0]
return &CodexModelsManifest{
Body: merged,
ETag: codexModelsManifestBodyETag(merged),
}, &firstAccount, nil
}
@@ -0,0 +1,82 @@
//go:build unit
package service
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
)
func TestMergeCodexModelsManifestBodiesUnionAndOrder(t *testing.T) {
first := `{"object":"codex.manifest","models":[{"slug":"model-a","display_name":"A from first"},{"slug":"model-b"}]}`
second := `{"object":"codex.manifest","models":[{"slug":"model-a","display_name":"A from second"},{"slug":"model-c"}]}`
merged, err := mergeCodexModelsManifestBodies([][]byte{[]byte(first), []byte(second)})
require.NoError(t, err)
var envelope map[string]json.RawMessage
require.NoError(t, json.Unmarshal(merged, &envelope))
var objectField string
require.NoError(t, json.Unmarshal(envelope["object"], &objectField))
require.Equal(t, "codex.manifest", objectField)
var models []map[string]any
require.NoError(t, json.Unmarshal(envelope["models"], &models))
require.Len(t, models, 3)
require.Equal(t, "model-a", models[0]["slug"])
require.Equal(t, "A from first", models[0]["display_name"], "重复 slug 必须取配置顺序靠前账号的条目")
require.Equal(t, "model-b", models[1]["slug"])
require.Equal(t, "model-c", models[2]["slug"])
}
func TestMergeCodexModelsManifestBodiesEnvelopeFromFirstBody(t *testing.T) {
first := `{"object":"codex.manifest","extra_field":"from-first","models":[{"slug":"model-a"}]}`
second := `{"object":"other","extra_field":"from-second","another":"only-in-second","models":[]}`
merged, err := mergeCodexModelsManifestBodies([][]byte{[]byte(first), []byte(second)})
require.NoError(t, err)
var envelope map[string]any
require.NoError(t, json.Unmarshal(merged, &envelope))
require.Equal(t, "codex.manifest", envelope["object"])
require.Equal(t, "from-first", envelope["extra_field"], "顶层字段以第一个信封为基底")
require.NotContains(t, envelope, "another")
}
func TestMergeCodexModelsManifestBodiesHandlesSluglessEntries(t *testing.T) {
first := `{"models":[{"slug":"model-a"},{"display_name":"no slug one"}]}`
second := `{"models":[{"display_name":"no slug one"},{"display_name":"no slug two"},{"slug":"model-a"}]}`
merged, err := mergeCodexModelsManifestBodies([][]byte{[]byte(first), []byte(second)})
require.NoError(t, err)
var envelope struct {
Models []map[string]any `json:"models"`
}
require.NoError(t, json.Unmarshal(merged, &envelope))
require.Len(t, envelope.Models, 3)
require.Equal(t, "model-a", envelope.Models[0]["slug"])
require.Equal(t, "no slug one", envelope.Models[1]["display_name"], "无 slug 条目按出现顺序保留一次")
require.Equal(t, "no slug two", envelope.Models[2]["display_name"])
}
func TestMergeCodexModelsManifestBodiesSingleBody(t *testing.T) {
body := `{"models":[{"slug":"model-a"}]}`
merged, err := mergeCodexModelsManifestBodies([][]byte{[]byte(body)})
require.NoError(t, err)
require.JSONEq(t, body, string(merged))
}
func TestMergeCodexModelsManifestBodiesRejectsInvalidInput(t *testing.T) {
_, err := mergeCodexModelsManifestBodies(nil)
require.Error(t, err)
_, err = mergeCodexModelsManifestBodies([][]byte{[]byte(`{`)})
require.Error(t, err)
_, err = mergeCodexModelsManifestBodies([][]byte{[]byte(`{}`), []byte(`{"models":{}`)})
require.Error(t, err, "models 非数组必须报错")
}
@@ -30,12 +30,20 @@ import (
var chatgptCodexModelsURL = "https://chatgpt.com/backend-api/codex/models"
const (
codexModelsManifestCacheBodyLimit = 1 << 20
codexModelsManifestCacheMaxEntries = 64
codexModelsManifestCacheTTL = 30 * time.Second
codexModelsManifestCacheStaleTTL = 5 * time.Minute
codexModelsManifestRequestTimeout = 15 * time.Second
codexAutoModelPrefix = "codex-auto-"
codexModelsManifestCacheBodyLimit = 1 << 20
// codexModelsManifestCacheMaxEntries 上限按「账号数 × 客户端版本数 × 代理形态」
// 估算:缓存同时覆盖 OAuth 与 API Key 账号,且缓存键含 Authorization 与
// Version 头,不同客户端版本各自占一条;64 条在大规模部署下会被淘汰导致
// 额外上游请求。单条清单通常几十 KB,512 条最坏内存占用在几十 MB 量级。
codexModelsManifestCacheMaxEntries = 512
// 三段时效:≤TTL 为新鲜(直接返回缓存,零上游请求);TTL 到 StaleTTL 之间
// 乐观返回旧值并后台单飞刷新(携带上游 ETag,304 续期);超过 StaleTTL 丢弃
// 缓存同步等待刷新。TTL 取 1 分钟:manifest 变化低频,1 分钟内同账号重复
// 请求完全吸收;StaleTTL 取 5 分钟,控制旧内容最长可见时间在分钟级。
codexModelsManifestCacheTTL = 60 * time.Second
codexModelsManifestCacheStaleTTL = 5 * time.Minute
codexModelsManifestRequestTimeout = 15 * time.Second
codexAutoModelPrefix = "codex-auto-"
)
// FilterCodexModelIDsForGroup removes dedicated media-generation models,
@@ -1568,30 +1576,35 @@ func (s *OpenAIGatewayService) FetchCodexModelsManifest(ctx context.Context, acc
useAPIKeyUpstream: useAPIKeyUpstream,
}
if useAPIKeyUpstream {
return s.fetchCachedAPIKeyCodexModelsManifest(ctx, request, ifNoneMatch)
return s.fetchCachedCodexModelsManifest(ctx, request, s.fetchCodexModelsManifestUpstreamForRequest(request), ifNoneMatch)
}
manifest, fetchErr := s.fetchCodexModelsManifestUpstream(ctx, request, ifNoneMatch)
if !credAccount.IsOpenAIAgentIdentity() || !isAgentIdentityTaskInvalidCodexModelsError(fetchErr) {
s.handleCodexModelsManifestAccountAuthError(ctx, account, credAccount, fetchErr)
return manifest, fetchErr
}
expectedTaskID := strings.TrimSpace(credAccount.GetCredential("task_id"))
if recoverErr := s.recoverAgentIdentityTask(ctx, credAccount, expectedTaskID); recoverErr != nil {
return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_AUTH_FAILED", "agent identity task recovery failed: %v", recoverErr)
}
authHeaders, authErr := s.buildOpenAIAuthenticationHeaders(ctx, credAccount, "")
if authErr != nil {
return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_AUTH_FAILED", "build Codex models authentication after task recovery: %v", authErr)
}
request.headers.Del("Authorization")
request.headers.Del("ChatGPT-Account-ID")
for key, values := range authHeaders {
for _, value := range values {
request.headers.Add(key, value)
// OAuth 账号同样经过账号级缓存;闭包保留 agent identity 任务恢复逻辑,
// 错误时仍交给 handleCodexModelsManifestAccountAuthError 处理账号状态。
oauthFetch := func(fetchCtx context.Context, ifNoneMatch string) (*CodexModelsManifest, error) {
manifest, fetchErr := s.fetchCodexModelsManifestUpstream(fetchCtx, request, ifNoneMatch)
if !credAccount.IsOpenAIAgentIdentity() || !isAgentIdentityTaskInvalidCodexModelsError(fetchErr) {
s.handleCodexModelsManifestAccountAuthError(fetchCtx, account, credAccount, fetchErr)
return manifest, fetchErr
}
expectedTaskID := strings.TrimSpace(credAccount.GetCredential("task_id"))
if recoverErr := s.recoverAgentIdentityTask(fetchCtx, credAccount, expectedTaskID); recoverErr != nil {
return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_AUTH_FAILED", "agent identity task recovery failed: %v", recoverErr)
}
authHeaders, authErr := s.buildOpenAIAuthenticationHeaders(fetchCtx, credAccount, "")
if authErr != nil {
return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_AUTH_FAILED", "build Codex models authentication after task recovery: %v", authErr)
}
request.headers.Del("Authorization")
request.headers.Del("ChatGPT-Account-ID")
for key, values := range authHeaders {
for _, value := range values {
request.headers.Add(key, value)
}
}
setOpenAIChatGPTAccountHeaders(request.headers, credAccount)
return s.fetchCodexModelsManifestUpstream(fetchCtx, request, ifNoneMatch)
}
setOpenAIChatGPTAccountHeaders(request.headers, credAccount)
return s.fetchCodexModelsManifestUpstream(ctx, request, ifNoneMatch)
return s.fetchCachedCodexModelsManifest(ctx, request, oauthFetch, ifNoneMatch)
}
func isAgentIdentityTaskInvalidCodexModelsError(err error) bool {
@@ -1631,7 +1644,7 @@ func (s *OpenAIGatewayService) handleCodexModelsManifestAccountAuthError(ctx con
s.handleOpenAIAccountUpstreamError(ctx, account, upstreamErr.statusCode, headers, upstreamErr.body)
}
func (s *OpenAIGatewayService) fetchCachedAPIKeyCodexModelsManifest(ctx context.Context, request codexModelsManifestRequest, ifNoneMatch string) (*CodexModelsManifest, error) {
func (s *OpenAIGatewayService) fetchCachedCodexModelsManifest(ctx context.Context, request codexModelsManifestRequest, fetch func(ctx context.Context, ifNoneMatch string) (*CodexModelsManifest, error), ifNoneMatch string) (*CodexModelsManifest, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
@@ -1640,7 +1653,7 @@ func (s *OpenAIGatewayService) fetchCachedAPIKeyCodexModelsManifest(ctx context.
if state == codexModelsManifestCacheFresh {
return codexModelsManifestForClient(manifest, ifNoneMatch), nil
}
resultCh := s.refreshCachedAPIKeyCodexModelsManifest(cacheKey, request)
resultCh := s.refreshCachedCodexModelsManifest(cacheKey, request, fetch)
if state == codexModelsManifestCacheStale {
return codexModelsManifestForClient(manifest, ifNoneMatch), nil
}
@@ -1659,14 +1672,16 @@ func (s *OpenAIGatewayService) fetchCachedAPIKeyCodexModelsManifest(ctx context.
}
}
func (s *OpenAIGatewayService) refreshCachedAPIKeyCodexModelsManifest(cacheKey string, request codexModelsManifestRequest) <-chan singleflight.Result {
func (s *OpenAIGatewayService) refreshCachedCodexModelsManifest(cacheKey string, request codexModelsManifestRequest, fetch func(ctx context.Context, ifNoneMatch string) (*CodexModelsManifest, error)) <-chan singleflight.Result {
return s.codexModelsManifestCache.refresh.DoChan(cacheKey, func() (any, error) {
ctx, cancel := context.WithTimeout(context.Background(), codexModelsManifestRequestTimeout)
defer cancel()
cached, _ := s.codexModelsManifestCache.get(cacheKey, time.Now())
ifNoneMatch := ""
if cached != nil {
ifNoneMatch = cached.upstreamETag
}
manifest, err := s.fetchCodexModelsManifestUpstream(context.Background(), request, ifNoneMatch)
manifest, err := fetch(ctx, ifNoneMatch)
if err != nil {
return nil, err
}
@@ -1681,6 +1696,12 @@ func (s *OpenAIGatewayService) refreshCachedAPIKeyCodexModelsManifest(cacheKey s
})
}
func (s *OpenAIGatewayService) fetchCodexModelsManifestUpstreamForRequest(request codexModelsManifestRequest) func(ctx context.Context, ifNoneMatch string) (*CodexModelsManifest, error) {
return func(ctx context.Context, ifNoneMatch string) (*CodexModelsManifest, error) {
return s.fetchCodexModelsManifestUpstream(ctx, request, ifNoneMatch)
}
}
func (s *OpenAIGatewayService) fetchCodexModelsManifestUpstream(ctx context.Context, request codexModelsManifestRequest, ifNoneMatch string) (*CodexModelsManifest, error) {
reqCtx, cancel := context.WithTimeout(ctx, codexModelsManifestRequestTimeout)
defer cancel()
@@ -1810,14 +1831,12 @@ func (s *OpenAIGatewayService) fetchCodexModelsManifestUpstream(ctx context.Cont
manifest := &CodexModelsManifest{
Body: body,
ETag: etag,
upstreamETag: etag,
upstreamSourceBody: append([]byte(nil), upstreamBody...),
convertedFromOpenAIModelList: convertedFromOpenAIModelList,
}
if request.useAPIKeyUpstream {
manifest.upstreamETag = etag
if !bytes.Equal(body, upstreamBody) {
manifest.ETag = codexModelsManifestBodyETag(body)
}
if request.useAPIKeyUpstream && !bytes.Equal(body, upstreamBody) {
manifest.ETag = codexModelsManifestBodyETag(body)
}
return manifest, nil
}
@@ -1737,11 +1737,12 @@ func TestFetchCodexModelsManifestDefaultClientVersion(t *testing.T) {
}
func TestFetchCodexModelsManifestNotModified(t *testing.T) {
var gotIfNoneMatch string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotIfNoneMatch = r.Header.Get("If-None-Match")
manifestBody := `{"models":[{"slug":"gpt-5.5"}]}`
var calls atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
calls.Add(1)
w.Header().Set("ETag", `W/"abc123"`)
w.WriteHeader(http.StatusNotModified)
_, _ = w.Write([]byte(manifestBody))
}))
defer server.Close()
@@ -1750,15 +1751,32 @@ func TestFetchCodexModelsManifestNotModified(t *testing.T) {
defer func() { chatgptCodexModelsURL = original }()
s := &OpenAIGatewayService{}
manifest, err := s.FetchCodexModelsManifest(context.Background(), newCodexModelsTestAccount(), "0.137.0", `W/"abc123"`)
account := newCodexModelsTestAccount()
first, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
if err != nil {
t.Fatalf("FetchCodexModelsManifest returned error: %v", err)
}
if !manifest.NotModified {
t.Error("expected NotModified to be true")
if first.NotModified {
t.Fatal("cold fetch must return the manifest body")
}
if gotIfNoneMatch != `W/"abc123"` {
t.Errorf("if-none-match header: got %q", gotIfNoneMatch)
if first.ETag != `W/"abc123"` {
t.Errorf("etag not passed through: got %q", first.ETag)
}
// OAuth 路径接入缓存后,客户端 If-None-Match 与缓存内容 ETag 比较,不再透传上游:
// 新鲜期内命中缓存且匹配直接 304,零上游请求。
second, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", `W/"abc123"`)
if err != nil {
t.Fatalf("cached fetch returned error: %v", err)
}
if !second.NotModified {
t.Fatal("matching cached ETag must return NotModified")
}
if second.ETag != `W/"abc123"` {
t.Errorf("NotModified etag: got %q", second.ETag)
}
if got := calls.Load(); got != 1 {
t.Errorf("upstream calls: got %d, want 1", got)
}
}
@@ -2746,22 +2764,22 @@ func TestFetchCodexModelsManifestAPIKeyCacheBoundsEntriesAndBodySize(t *testing.
t.Fatalf("body-size bounded cache calls: got %d, want 5", got)
}
for i := int64(10); i < 75; i++ {
for i := int64(10); i < 523; i++ {
account := newCodexModelsAPIKeyTestAccount("https://bounded.example")
account.ID = i
fetch(account)
}
last := newCodexModelsAPIKeyTestAccount("https://bounded.example")
last.ID = 74
last.ID = 522
fetch(last)
if got := calls.Load(); got != 70 {
t.Fatalf("most recent cache entry was not retained: calls=%d, want 70", got)
if got := calls.Load(); got != 518 {
t.Fatalf("most recent cache entry was not retained: calls=%d, want 518", got)
}
first := newCodexModelsAPIKeyTestAccount("https://bounded.example")
first.ID = 10
fetch(first)
if got := calls.Load(); got != 71 {
t.Errorf("oldest cache entry was not evicted: calls=%d, want 71", got)
if got := calls.Load(); got != 519 {
t.Errorf("oldest cache entry was not evicted: calls=%d, want 519", got)
}
}
@@ -3255,3 +3273,271 @@ func TestFetchCodexModelsManifestAPIKeyUsesOfficialOpenAIModelsEndpoint(t *testi
})
}
}
// --- OAuth manifest 账号级缓存(统一缓存策略) ---
// newCodexModelsOAuthCacheServer 返回一个记录调用次数、返回给定 manifest 的
// ChatGPT Codex models 测试上游。
func newCodexModelsOAuthCacheServer(t *testing.T, body string) (*httptest.Server, *atomic.Int32) {
t.Helper()
calls := &atomic.Int32{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
calls.Add(1)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(body))
}))
t.Cleanup(server.Close)
original := chatgptCodexModelsURL
chatgptCodexModelsURL = server.URL
t.Cleanup(func() { chatgptCodexModelsURL = original })
return server, calls
}
func expireCodexModelsManifestCache(s *OpenAIGatewayService, age time.Duration) {
s.codexModelsManifestCache.mu.Lock()
for key, entry := range s.codexModelsManifestCache.entries {
entry.expiresAt = time.Now().Add(-age)
entry.staleUntil = time.Now().Add(codexModelsManifestCacheStaleTTL - age)
s.codexModelsManifestCache.entries[key] = entry
}
s.codexModelsManifestCache.mu.Unlock()
}
func TestFetchCodexModelsManifestOAuthFreshWindowZeroUpstreamRequests(t *testing.T) {
_, calls := newCodexModelsOAuthCacheServer(t, `{"models":[{"slug":"gpt-5.5"}]}`)
s := &OpenAIGatewayService{}
account := newCodexModelsTestAccount()
for i := 0; i < 10; i++ {
manifest, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
require.False(t, manifest.NotModified)
require.JSONEq(t, `{"models":[{"slug":"gpt-5.5"}]}`, string(manifest.Body))
}
require.EqualValues(t, 1, calls.Load(), "新鲜期内重复请求不得打到 chatgpt.com")
}
func TestFetchCodexModelsManifestOAuthStaleServesOldValueAndRefreshesInBackground(t *testing.T) {
calls := &atomic.Int32{}
refreshStarted := make(chan struct{})
var refreshOnce sync.Once
release := make(chan struct{})
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
if calls.Add(1) == 2 {
refreshOnce.Do(func() { close(refreshStarted) })
<-release
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"models":[{"slug":"new"}]}`))
}))
t.Cleanup(server.Close)
original := chatgptCodexModelsURL
chatgptCodexModelsURL = server.URL
t.Cleanup(func() { chatgptCodexModelsURL = original })
s := &OpenAIGatewayService{}
account := newCodexModelsTestAccount()
first, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
require.Contains(t, string(first.Body), `"new"`)
expireCodexModelsManifestCache(s, 2*time.Minute)
resultCh := make(chan *CodexModelsManifest, 1)
errCh := make(chan error, 1)
go func() {
manifest, fetchErr := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
resultCh <- manifest
errCh <- fetchErr
}()
select {
case <-refreshStarted:
case <-time.After(time.Second):
t.Fatal("乐观期后台刷新未触发")
}
// 乐观期:立即返回旧值,不同步等待刷新。
select {
case manifest := <-resultCh:
require.NoError(t, <-errCh)
require.Contains(t, string(manifest.Body), `"new"`)
case <-time.After(100 * time.Millisecond):
t.Fatal("乐观期必须立即返回缓存旧值")
}
require.EqualValues(t, 2, calls.Load())
close(release)
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
manifest, fetchErr := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
if fetchErr == nil && manifest != nil && !manifest.NotModified {
require.Contains(t, string(manifest.Body), `"new"`)
}
s.codexModelsManifestCache.mu.Lock()
fresh := len(s.codexModelsManifestCache.entries) > 0
s.codexModelsManifestCache.mu.Unlock()
if fresh {
break
}
time.Sleep(10 * time.Millisecond)
}
require.EqualValues(t, 2, calls.Load(), "后台刷新必须单飞")
}
func TestFetchCodexModelsManifestOAuthOverdueSynchronousRefreshAndFailure(t *testing.T) {
t.Run("overdue refresh waits and re-caches", func(t *testing.T) {
calls := &atomic.Int32{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
if calls.Add(1) == 2 {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"models":[{"slug":"refreshed"}]}`))
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"models":[{"slug":"stale"}]}`))
}))
t.Cleanup(server.Close)
original := chatgptCodexModelsURL
chatgptCodexModelsURL = server.URL
t.Cleanup(func() { chatgptCodexModelsURL = original })
s := &OpenAIGatewayService{}
account := newCodexModelsTestAccount()
_, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
// 超过 5 分钟:缓存条目被丢弃,同步等待上游刷新。
expireCodexModelsManifestCache(s, 6*time.Minute)
manifest, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
require.Contains(t, string(manifest.Body), `"refreshed"`)
require.EqualValues(t, 2, calls.Load())
// 新内容写回缓存并重新开始新鲜期。
again, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
require.Contains(t, string(again.Body), `"refreshed"`)
require.EqualValues(t, 2, calls.Load())
})
t.Run("overdue upstream failure returns error not stale cache", func(t *testing.T) {
fail := &atomic.Bool{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
if fail.Load() {
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(`{"detail":"boom"}`))
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"models":[{"slug":"stale"}]}`))
}))
t.Cleanup(server.Close)
original := chatgptCodexModelsURL
chatgptCodexModelsURL = server.URL
t.Cleanup(func() { chatgptCodexModelsURL = original })
s := &OpenAIGatewayService{}
account := newCodexModelsTestAccount()
_, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
fail.Store(true)
expireCodexModelsManifestCache(s, 6*time.Minute)
_, err = s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.Error(t, err, "超期后上游失败必须返回错误而非旧缓存")
require.Equal(t, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", infraerrors.Reason(err))
})
}
func TestFetchCodexModelsManifestOAuthTokenChangeCacheMiss(t *testing.T) {
_, calls := newCodexModelsOAuthCacheServer(t, `{"models":[]}`)
s := &OpenAIGatewayService{}
account := newCodexModelsTestAccount()
_, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
require.EqualValues(t, 1, calls.Load())
// 令牌刷新后缓存键变化,下一次拉取视为未命中并同步请求上游。
account.Credentials["access_token"] = "rotated-access-token"
manifest, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
require.NoError(t, err)
require.Contains(t, string(manifest.Body), `"models"`)
require.EqualValues(t, 2, calls.Load())
}
func TestFetchCodexModelsManifestOAuthSharedAcrossGroupsWithIndependentFiltering(t *testing.T) {
started := make(chan struct{})
var startedOnce sync.Once
release := make(chan struct{})
calls := &atomic.Int32{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
calls.Add(1)
startedOnce.Do(func() { close(started) })
<-release
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"models":[{"slug":"model-a"},{"slug":"model-b"}]}`))
}))
t.Cleanup(server.Close)
original := chatgptCodexModelsURL
chatgptCodexModelsURL = server.URL
t.Cleanup(func() { chatgptCodexModelsURL = original })
s := &OpenAIGatewayService{
accountRepo: codexModelsVisibilityAccountRepo{
byGroup: map[int64][]Account{
91: {},
92: {},
},
},
}
account := newCodexModelsTestAccount()
groupA := &Group{ID: 91, Platform: PlatformOpenAI, ModelsListConfig: GroupModelsListConfig{Enabled: true, Models: []string{"model-a"}}}
groupB := &Group{ID: 92, Platform: PlatformOpenAI, ModelsListConfig: GroupModelsListConfig{Enabled: true, Models: []string{"model-b"}}}
begin := make(chan struct{})
type result struct {
slugs []string
err error
}
results := make(chan result, 2)
for _, group := range []*Group{groupA, groupB} {
go func(g *Group) {
<-begin
manifest, err := s.FetchCodexModelsManifest(context.Background(), account, "0.137.0", "")
if err == nil {
err = s.MergeGroupConfiguredCodexModels(context.Background(), g, manifest, "")
}
slugs := []string{}
if err == nil {
slugs = codexManifestModelSlugs(t, manifest.Body)
}
results <- result{slugs: slugs, err: err}
}(group)
}
close(begin)
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("upstream request did not start")
}
close(release)
got := map[int64][]string{91: nil, 92: nil}
for i := 0; i < 2; i++ {
r := <-results
require.NoError(t, r.err)
if len(r.slugs) == 1 && r.slugs[0] == "model-a" {
got[91] = r.slugs
} else if len(r.slugs) == 1 && r.slugs[0] == "model-b" {
got[92] = r.slugs
} else {
t.Fatalf("unexpected filtered slugs: %v", r.slugs)
}
}
require.Equal(t, []string{"model-a"}, got[91])
require.Equal(t, []string{"model-b"}, got[92])
require.EqualValues(t, 1, calls.Load(), "同一账号两个分组同时请求时只发一次上游请求")
}