fix: integrate account tests with Codex overdraft
This commit is contained in:
1 parent
43391b6bb0
commit
f614c3e989
10 files changed
+280
-28
No files matched your search
@@ -20,10 +20,11 @@
|
||||
- 支持 `/v1/responses`、转换为 Responses 的 `/v1/chat/completions` 和 `/v1/messages`,以及 Responses WebSocket v2。
|
||||
- 在适用请求的最后一条用户消息后注入一对无操作工具调用,用于保持与参考实现一致的请求形态。
|
||||
- 当 5h 或 7d 额度首次达到 100%,或收到带明确额度证据的 429 时,最多执行 5 次真实探测。
|
||||
- 管理页面的 OpenAI OAuth 常规文本“测试账号连接”也使用相同请求形态,并接入同一套额度探测状态机。
|
||||
- 分别维护 5h、7d 透支周期,并在管理页面显示状态、请求数、Token、账号金额和预计恢复时间。
|
||||
- 将探测状态保存在现有 `accounts.extra` JSONB 字段中,不新增数据库表,不需要手动执行数据库迁移(migration)。
|
||||
|
||||
以下端点不启用透支:`/responses/compact`、图片生成、Embedding、Count Tokens 和 Live。
|
||||
以下端点不启用透支:`/responses/compact`、图片生成、Embedding、Count Tokens 和 Live;API Key、Shadow、图片及 Compact 的后台测试也不启用透支。
|
||||
|
||||
## 前置条件
|
||||
|
||||
@@ -255,11 +256,11 @@ docker compose \
|
||||
|
||||
### 第三层:查看真实探测日志
|
||||
|
||||
只有账号 5h/7d 额度达到 100%,或者收到带明确额度信息的上游 429,才会启动探测。额度未耗尽时没有相关日志是正常现象。
|
||||
只有账号 5h/7d 额度达到 100%,或者收到带明确额度信息的上游 429,才会启动探测。额度达到 100% 后,可以在管理页面对该 OpenAI OAuth 账号执行常规文本“测试账号连接”;测试请求本身会使用透支请求形态,并触发或接续同一额度周期的探测。额度未耗尽时没有相关日志是正常现象。
|
||||
|
||||
```bash
|
||||
docker logs --since 30m sub2api 2>&1 | \
|
||||
grep -E 'codex_quota_overdraft_(probe|state|pause)'
|
||||
grep -E 'codex_quota_overdraft_(probe|state|pause|stale_rate_limit)'
|
||||
```
|
||||
|
||||
关键日志:
|
||||
@@ -271,6 +272,7 @@ docker logs --since 30m sub2api 2>&1 | \
|
||||
| `codex_quota_overdraft_probe_inconclusive` | 网络、超时、5xx 等导致无法确认,约 1 分钟后可重试 |
|
||||
| `codex_quota_overdraft_probe_claim_failed` | 无法在 PostgreSQL 中原子领取探测任务,需要排查数据库或版本 |
|
||||
| `codex_quota_overdraft_state_persist_failed` | 状态无法写入 `accounts.extra`,页面可能不显示最新结果 |
|
||||
| `codex_quota_overdraft_stale_rate_limit_cleared` | 探测已证明账号可用,并清除了并发 429 遗留的账号级限流状态 |
|
||||
|
||||
真正“透支成功”的最直接证据是:
|
||||
|
||||
@@ -467,7 +469,9 @@ docker compose \
|
||||
|
||||
### 页面仍显示“限流中”
|
||||
|
||||
本功能只绕过 5h/7d 额度阈值产生的预暂停。以下情况仍会正常限流或暂停:账号禁用、401/403、5 次探测均确认额度耗尽、代理/网络错误、模型级冷却、并发限制或其他上游 429。先按 `account_id` 检查应用日志和数据库中的 `reason_code`。
|
||||
页面的“限流中”直接来自账号记录的 `rate_limit_reset_at`。旧实现存在竞态:探测开始后若另一个 429 更新了该时间,即使探测最终 `passed`,严格时间比较也可能不清理它,因此出现“实际透支可用但页面限流中”。当前版本在 `passed/recovered` 时会清除该残留状态,并记录 `codex_quota_overdraft_stale_rate_limit_cleared`;`failed` 不会清理。
|
||||
|
||||
其他真实限制仍会正常生效,包括账号禁用、401/403、5 次探测均确认额度耗尽、代理/网络错误、模型级冷却、并发限制或没有明确额度证据的其他上游 429。若状态持续存在,先按 `account_id` 检查上述日志、数据库中的 `extra->'codex_quota_overdraft_probe'` 和 `rate_limit_reset_at`,区分额度透支与普通限流。
|
||||
|
||||
### `pq: could not determine data type of parameter $1`
|
||||
|
||||
|
||||
@@ -10,6 +10,12 @@
|
||||
- 功能开关:`gateway.codex_quota_overdraft_enabled`
|
||||
- 代码默认值:关闭;`deploy/config.example.yaml` 部署示例默认开启
|
||||
|
||||
## 后台测试连接与残留限流修复
|
||||
|
||||
管理页面的 OpenAI OAuth 常规文本“测试账号连接”使用与真实业务请求相同的透支请求形态。测试成功并取得 5h/7d 快照后会触发额度观察;测试返回带明确额度证据的 429 时交给透支协调器执行最多 5 次真实探测,不再直接写入普通账号级限流。API Key、Shadow、图片和 Compact 测试不启用该行为。
|
||||
|
||||
探测得到 `passed`,或额度周期进入 `recovered` 后,会清除仍残留的账号级 `rate_limit_reset_at` 以及本功能或额度阈值产生的暂停。这里不再要求当前重置时间与探测开始时记录的 `observed_rate_limit_reset_at` 完全相同,因为并发 429 可能在探测期间更新该字段;`failed` 状态不会清理限流。发生清理时记录 `codex_quota_overdraft_stale_rate_limit_cleared`。
|
||||
|
||||
## PostgreSQL 兼容性修复
|
||||
|
||||
探测计划使用原子 claim 写入 `accounts.extra`。`jsonb_build_object` 的键参数必须显式
|
||||
@@ -39,6 +45,7 @@ gateway:
|
||||
- `/v1/chat/completions` 转换到 Responses 的请求
|
||||
- `/v1/messages` 转换到 Responses 的请求
|
||||
- Responses WebSocket v2 首轮和后续轮
|
||||
- 管理页面 OpenAI OAuth 常规文本“测试账号连接”
|
||||
|
||||
每次最终发往上游前,如果 `input` 最后一项是用户消息,则追加一对与参考项目相同的无操作 `custom_tool_call` 和 `custom_tool_call_output`。请求体超过 32 MiB、JSON 无效、形状不匹配或已经注入时保持原样。
|
||||
|
||||
@@ -85,6 +92,9 @@ gateway:
|
||||
- `backend/internal/service/openai_ws_v2_passthrough_adapter.go`
|
||||
- `backend/internal/service/ratelimit_service.go`
|
||||
- `backend/internal/service/account_usage_service.go`
|
||||
- `backend/internal/service/account_test_service.go`
|
||||
- `backend/internal/service/account_test_service_openai_test.go`
|
||||
- `backend/internal/service/account_test_service_openai_compact_test.go`
|
||||
- `backend/internal/service/wire.go`
|
||||
- `backend/cmd/server/wire_gen.go`
|
||||
|
||||
@@ -115,7 +125,7 @@ gateway:
|
||||
cd backend
|
||||
gofmt -w internal/service/openai_codex_quota_overdraft*.go
|
||||
go test -tags=unit ./internal/repository ./internal/handler ./internal/service \
|
||||
-run 'CodexQuotaOverdraft|AccountSchedulingThreshold|SchedulerSnapshot|SchedulableAccountQueryScopesCodex' -count=1
|
||||
-run 'CodexQuotaOverdraft|AccountTestService_OpenAI|AccountSchedulingThreshold|SchedulerSnapshot|SchedulableAccountQueryScopesCodex' -count=1
|
||||
go test ./internal/repository ./internal/handler ./internal/service ./cmd/server -run '^$' -count=1
|
||||
|
||||
cd ../frontend
|
||||
@@ -135,6 +145,7 @@ git diff --check
|
||||
codex_quota_overdraft_probe_passed
|
||||
codex_quota_overdraft_probe_failed
|
||||
codex_quota_overdraft_probe_inconclusive
|
||||
codex_quota_overdraft_stale_rate_limit_cleared
|
||||
```
|
||||
|
||||
数据库状态可这样检查:
|
||||
|
||||
+2
-2
@@ -64,10 +64,10 @@ gateway:
|
||||
|
||||
```bash
|
||||
docker logs --since 30m sub2api 2>&1 | \
|
||||
grep -E 'codex_quota_overdraft_(probe|state|pause)'
|
||||
grep -E 'codex_quota_overdraft_(probe|state|pause|stale_rate_limit)'
|
||||
```
|
||||
|
||||
出现 `codex_quota_overdraft_probe_passed`,页面显示“透支中”,并且后续真实业务请求成功,即可确认透支功能完整生效。额度未达到 100% 时没有探测日志是正常现象。
|
||||
出现 `codex_quota_overdraft_probe_passed`,页面显示“透支中”,并且后续真实业务请求成功,即可确认透支功能完整生效。OpenAI OAuth 常规文本“测试账号连接”也会使用透支请求形态并接入同一探测状态机;API Key、Shadow、图片和 Compact 测试除外。额度未达到 100% 时没有探测日志是正常现象。
|
||||
|
||||
## 来源、许可证与风险
|
||||
|
||||
|
||||
@@ -201,7 +201,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
|
||||
usageCache := service.NewUsageCache()
|
||||
codexQuotaOverdraftCoordinator := service.ProvideCodexQuotaOverdraftCoordinator(accountRepository, httpUpstream, openAITokenProvider, tlsFingerprintProfileService, configConfig, tempUnschedCache, openAIGatewayService, rateLimitService)
|
||||
accountUsageService := service.ProvideAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService, openAIGatewayService, codexQuotaOverdraftCoordinator)
|
||||
accountTestService := service.ProvideAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService, openAIGatewayService, settingService)
|
||||
accountTestService := service.ProvideAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService, openAIGatewayService, settingService, codexQuotaOverdraftCoordinator)
|
||||
crsSyncService := service.NewCRSSyncService(accountRepository, proxyRepository, oAuthService, openAIOAuthService, geminiOAuthService, configConfig)
|
||||
accountHandler := admin.ProvideAccountHandler(adminService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, grokOAuthService, rateLimitService, accountUsageService, accountTestService, concurrencyService, crsSyncService, sessionLimitCache, rpmCache, compositeTokenCacheInvalidator, grokQuotaService)
|
||||
adminAnnouncementHandler := admin.NewAnnouncementHandler(announcementService)
|
||||
|
||||
@@ -146,6 +146,7 @@ type AccountTestService struct {
|
||||
cfg *config.Config
|
||||
settingService *SettingService
|
||||
tlsFPProfileService *TLSFingerprintProfileService
|
||||
codexQuotaOverdraft codexQuotaOverdraftAccountTestCoordinator
|
||||
agentIdentityTaskMu sync.Mutex
|
||||
agentIdentityWS agentIdentityWSConnectionInvalidator
|
||||
// grokWSDialer is optional; realtime account tests use the default OpenAI-style
|
||||
@@ -153,12 +154,23 @@ type AccountTestService struct {
|
||||
grokWSDialer openAIWSClientDialer
|
||||
}
|
||||
|
||||
type codexQuotaOverdraftAccountTestCoordinator interface {
|
||||
ObserveAccount(account *Account, preferredModel string)
|
||||
HandleQuota429(ctx context.Context, account *Account, headers http.Header, body []byte, preferredModel string) bool
|
||||
}
|
||||
|
||||
func (s *AccountTestService) SetSettingService(settingService *SettingService) {
|
||||
if s != nil {
|
||||
s.settingService = settingService
|
||||
}
|
||||
}
|
||||
|
||||
func (s *AccountTestService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) {
|
||||
if s != nil {
|
||||
s.codexQuotaOverdraft = coordinator
|
||||
}
|
||||
}
|
||||
|
||||
// NewAccountTestService creates a new AccountTestService
|
||||
func NewAccountTestService(
|
||||
accountRepo AccountRepository,
|
||||
@@ -693,6 +705,11 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
|
||||
}
|
||||
payload := createOpenAITestPayload(upstreamTestModelID, isOAuth)
|
||||
payloadBytes, _ := json.Marshal(payload)
|
||||
if s.codexQuotaOverdraftTestEnabled(account) {
|
||||
if updated, changed, err := injectCodexQuotaOverdraft(payloadBytes); err == nil && changed {
|
||||
payloadBytes = updated
|
||||
}
|
||||
}
|
||||
|
||||
// Send test_start event once. A task-invalid Agent Identity response may
|
||||
// restart this probe after registering a replacement task.
|
||||
@@ -776,7 +793,11 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
|
||||
return s.testOpenAIAccountConnection(c, account, modelID, prompt, mode)
|
||||
}
|
||||
if resp.StatusCode == http.StatusTooManyRequests {
|
||||
s.reconcileOpenAI429State(ctx, account, resp.Header, body)
|
||||
handled := s.codexQuotaOverdraft != nil && s.codexQuotaOverdraftTestEnabled(account) &&
|
||||
s.codexQuotaOverdraft.HandleQuota429(ctx, account, resp.Header, body, upstreamTestModelID)
|
||||
if !handled {
|
||||
s.reconcileOpenAI429State(ctx, account, resp.Header, body)
|
||||
}
|
||||
}
|
||||
// 401 Unauthorized: 标记账号为永久错误
|
||||
if resp.StatusCode == http.StatusUnauthorized && s.accountRepo != nil {
|
||||
@@ -785,9 +806,19 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
|
||||
}
|
||||
return s.sendErrorAndEnd(c, fmt.Sprintf("API returned %d: %s", resp.StatusCode, string(body)))
|
||||
}
|
||||
|
||||
// Process SSE stream
|
||||
return s.processOpenAIStream(c, resp.Body)
|
||||
if err := s.processOpenAIStream(c, resp.Body); err != nil {
|
||||
return err
|
||||
}
|
||||
if s.codexQuotaOverdraft != nil && s.codexQuotaOverdraftTestEnabled(account) {
|
||||
s.codexQuotaOverdraft.ObserveAccount(account, upstreamTestModelID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *AccountTestService) codexQuotaOverdraftTestEnabled(account *Account) bool {
|
||||
return s != nil && s.cfg != nil && s.cfg.Gateway.CodexQuotaOverdraftEnabled &&
|
||||
isCodexQuotaOverdraftAccount(account)
|
||||
}
|
||||
|
||||
// testGrokAccountConnection routes Grok admin connectivity tests by explicit mode first,
|
||||
@@ -2548,6 +2579,7 @@ func createOpenAITestPayload(modelID string, isOAuth bool) map[string]any {
|
||||
"model": modelID,
|
||||
"input": []map[string]any{
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": []map[string]any{
|
||||
{
|
||||
|
||||
@@ -44,6 +44,7 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactOAuthSuccessPersi
|
||||
svc := &AccountTestService{
|
||||
accountRepo: repo,
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}},
|
||||
}
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
@@ -63,6 +64,7 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactOAuthSuccessPersi
|
||||
require.Equal(t, "chatgpt-acc", upstream.lastReq.Header.Get("chatgpt-account-id"))
|
||||
require.Equal(t, "true", upstream.lastReq.Header.Get("x-openai-fedramp"))
|
||||
require.Equal(t, "gpt-5.4", gjson.GetBytes(upstream.lastBody, "model").String())
|
||||
require.False(t, gjson.GetBytes(upstream.lastBody, "input.1").Exists())
|
||||
|
||||
updates := <-updateCalls
|
||||
require.Equal(t, true, updates["openai_compact_supported"])
|
||||
|
||||
@@ -73,6 +73,35 @@ type openAIAccountTestRepo struct {
|
||||
setErrorMsg string
|
||||
}
|
||||
|
||||
type accountTestOverdraftCoordinatorStub struct {
|
||||
observeCalls int
|
||||
observedAccount *Account
|
||||
observedModel string
|
||||
handleCalls int
|
||||
handledAccount *Account
|
||||
handledModel string
|
||||
handleResult bool
|
||||
}
|
||||
|
||||
func (s *accountTestOverdraftCoordinatorStub) ObserveAccount(account *Account, preferredModel string) {
|
||||
s.observeCalls++
|
||||
s.observedAccount = account
|
||||
s.observedModel = preferredModel
|
||||
}
|
||||
|
||||
func (s *accountTestOverdraftCoordinatorStub) HandleQuota429(
|
||||
_ context.Context,
|
||||
account *Account,
|
||||
_ http.Header,
|
||||
_ []byte,
|
||||
preferredModel string,
|
||||
) bool {
|
||||
s.handleCalls++
|
||||
s.handledAccount = account
|
||||
s.handledModel = preferredModel
|
||||
return s.handleResult
|
||||
}
|
||||
|
||||
func (r *openAIAccountTestRepo) UpdateExtra(_ context.Context, _ int64, updates map[string]any) error {
|
||||
r.updatedExtra = updates
|
||||
return nil
|
||||
@@ -137,6 +166,47 @@ func TestAccountTestService_OpenAISuccessPersistsSnapshotFromHeaders(t *testing.
|
||||
require.Contains(t, recorder.Body.String(), "test_complete")
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAIOAuthOverdraftTestInjectsAndObservesSnapshot(t *testing.T) {
|
||||
ctx, _ := newTestContext()
|
||||
|
||||
resp := newJSONResponse(http.StatusOK, "")
|
||||
resp.Body = io.NopCloser(strings.NewReader("data: {\"type\":\"response.completed\"}\n\n"))
|
||||
resp.Header.Set("x-codex-secondary-used-percent", "100")
|
||||
resp.Header.Set("x-codex-secondary-reset-after-seconds", "18000")
|
||||
resp.Header.Set("x-codex-secondary-window-minutes", "300")
|
||||
|
||||
repo := &openAIAccountTestRepo{}
|
||||
upstream := &queuedHTTPUpstream{responses: []*http.Response{resp}}
|
||||
coordinator := &accountTestOverdraftCoordinatorStub{}
|
||||
svc := &AccountTestService{
|
||||
accountRepo: repo,
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}},
|
||||
codexQuotaOverdraft: coordinator,
|
||||
}
|
||||
account := &Account{
|
||||
ID: 96,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeOAuth,
|
||||
Concurrency: 1,
|
||||
Credentials: map[string]any{"access_token": "test-token"},
|
||||
}
|
||||
|
||||
err := svc.testOpenAIAccountConnection(ctx, account, "gpt-5.4", "", "")
|
||||
require.NoError(t, err)
|
||||
require.Len(t, upstream.requests, 1)
|
||||
body, err := io.ReadAll(upstream.requests[0].Body)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "message", gjson.GetBytes(body, "input.0.type").String())
|
||||
require.Equal(t, "custom_tool_call", gjson.GetBytes(body, "input.1.type").String())
|
||||
require.Equal(t, "custom_tool_call_output", gjson.GetBytes(body, "input.2.type").String())
|
||||
require.Equal(t, gjson.GetBytes(body, "input.1.call_id").String(), gjson.GetBytes(body, "input.2.call_id").String())
|
||||
require.Equal(t, 1, coordinator.observeCalls)
|
||||
require.Same(t, account, coordinator.observedAccount)
|
||||
require.Equal(t, "gpt-5.4", coordinator.observedModel)
|
||||
require.Equal(t, 100.0, account.Extra["codex_5h_used_percent"])
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAIOAuthTestNormalizesGPT56Alias(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
ctx, _ := newTestContext()
|
||||
@@ -163,6 +233,8 @@ func TestAccountTestService_OpenAIOAuthTestNormalizesGPT56Alias(t *testing.T) {
|
||||
body, err := io.ReadAll(upstream.requests[0].Body)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "gpt-5.6-sol", gjson.GetBytes(body, "model").String())
|
||||
require.Equal(t, "message", gjson.GetBytes(body, "input.0.type").String())
|
||||
require.False(t, gjson.GetBytes(body, "input.1").Exists())
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAIShadowUsesParentCredentialsAndShadowModel(t *testing.T) {
|
||||
@@ -209,7 +281,13 @@ func TestAccountTestService_OpenAIShadowUsesParentCredentialsAndShadowModel(t *t
|
||||
},
|
||||
}
|
||||
upstream := &queuedHTTPUpstream{responses: []*http.Response{resp}}
|
||||
svc := &AccountTestService{accountRepo: repo, httpUpstream: upstream}
|
||||
coordinator := &accountTestOverdraftCoordinatorStub{}
|
||||
svc := &AccountTestService{
|
||||
accountRepo: repo,
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}},
|
||||
codexQuotaOverdraft: coordinator,
|
||||
}
|
||||
|
||||
err := svc.TestAccountConnection(ctx, shadow.ID, "gpt-5.3-codex-spark", "", "")
|
||||
require.NoError(t, err)
|
||||
@@ -220,6 +298,9 @@ func TestAccountTestService_OpenAIShadowUsesParentCredentialsAndShadowModel(t *t
|
||||
body, err := io.ReadAll(req.Body)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "gpt-5.3-codex-spark", gjson.GetBytes(body, "model").String())
|
||||
require.False(t, gjson.GetBytes(body, "input.1").Exists())
|
||||
require.Zero(t, coordinator.observeCalls)
|
||||
require.Zero(t, coordinator.handleCalls)
|
||||
require.Contains(t, recorder.Body.String(), `"success":true`)
|
||||
}
|
||||
|
||||
@@ -284,6 +365,72 @@ func TestAccountTestService_OpenAI429PersistsSnapshotAndRateLimitState(t *testin
|
||||
require.NotNil(t, account.RateLimitResetAt)
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAIQuota429UsesOverdraftCoordinatorWithoutRateLimitWrite(t *testing.T) {
|
||||
ctx, _ := newTestContext()
|
||||
|
||||
resp := newJSONResponse(http.StatusTooManyRequests, `{"error":{"type":"usage_limit_reached","message":"limit reached","resets_at":1777283883}}`)
|
||||
resp.Header.Set("x-codex-secondary-used-percent", "100")
|
||||
resp.Header.Set("x-codex-secondary-reset-after-seconds", "18000")
|
||||
resp.Header.Set("x-codex-secondary-window-minutes", "300")
|
||||
|
||||
repo := &openAIAccountTestRepo{}
|
||||
upstream := &queuedHTTPUpstream{responses: []*http.Response{resp}}
|
||||
coordinator := &accountTestOverdraftCoordinatorStub{handleResult: true}
|
||||
svc := &AccountTestService{
|
||||
accountRepo: repo,
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}},
|
||||
codexQuotaOverdraft: coordinator,
|
||||
}
|
||||
account := &Account{
|
||||
ID: 97,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeOAuth,
|
||||
Status: StatusActive,
|
||||
Concurrency: 1,
|
||||
Credentials: map[string]any{"access_token": "test-token"},
|
||||
}
|
||||
|
||||
err := svc.testOpenAIAccountConnection(ctx, account, "gpt-5.4", "", "")
|
||||
require.Error(t, err)
|
||||
require.Equal(t, 1, coordinator.handleCalls)
|
||||
require.Same(t, account, coordinator.handledAccount)
|
||||
require.Equal(t, "gpt-5.4", coordinator.handledModel)
|
||||
require.Zero(t, repo.rateLimitedID)
|
||||
require.Nil(t, repo.rateLimitedAt)
|
||||
require.Nil(t, account.RateLimitResetAt)
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAI429FallsBackWhenOverdraftDoesNotHandle(t *testing.T) {
|
||||
ctx, _ := newTestContext()
|
||||
|
||||
resp := newJSONResponse(http.StatusTooManyRequests, `{"error":{"type":"rate_limit_exceeded","message":"too many requests","resets_in_seconds":60}}`)
|
||||
repo := &openAIAccountTestRepo{}
|
||||
upstream := &queuedHTTPUpstream{responses: []*http.Response{resp}}
|
||||
coordinator := &accountTestOverdraftCoordinatorStub{}
|
||||
svc := &AccountTestService{
|
||||
accountRepo: repo,
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}},
|
||||
codexQuotaOverdraft: coordinator,
|
||||
}
|
||||
account := &Account{
|
||||
ID: 98,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeOAuth,
|
||||
Status: StatusActive,
|
||||
Concurrency: 1,
|
||||
Credentials: map[string]any{"access_token": "test-token"},
|
||||
}
|
||||
|
||||
err := svc.testOpenAIAccountConnection(ctx, account, "gpt-5.4", "", "")
|
||||
require.Error(t, err)
|
||||
require.Equal(t, 1, coordinator.handleCalls)
|
||||
require.Equal(t, account.ID, repo.rateLimitedID)
|
||||
require.NotNil(t, repo.rateLimitedAt)
|
||||
require.NotNil(t, account.RateLimitResetAt)
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAI429BodyOnlyPersistsRateLimitAndClearsStaleError(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
ctx, _ := newTestContext()
|
||||
@@ -432,8 +579,9 @@ func TestAccountTestService_OpenAIAPIKeyResponsesUsesCodexProbeHeaders(t *testin
|
||||
resp.Body = io.NopCloser(strings.NewReader("data: {\"type\":\"response.completed\"}\n\n"))
|
||||
upstream := &queuedHTTPUpstream{responses: []*http.Response{resp}}
|
||||
svc := &AccountTestService{
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{Enabled: false}}},
|
||||
httpUpstream: upstream,
|
||||
cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}, Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{Enabled: false}}},
|
||||
codexQuotaOverdraft: &accountTestOverdraftCoordinatorStub{},
|
||||
}
|
||||
account := &Account{
|
||||
ID: 95,
|
||||
@@ -453,6 +601,10 @@ func TestAccountTestService_OpenAIAPIKeyResponsesUsesCodexProbeHeaders(t *testin
|
||||
req := upstream.requests[0]
|
||||
require.Equal(t, "https://compat-upstream.example/v1/responses", req.URL.String())
|
||||
requireOpenAICodexProbeHeaders(t, req.Header)
|
||||
body, err := io.ReadAll(req.Body)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "message", gjson.GetBytes(body, "input.0.type").String())
|
||||
require.False(t, gjson.GetBytes(body, "input.1").Exists())
|
||||
}
|
||||
|
||||
func TestAccountTestService_OpenAIAPIKeyResponsesUnsupportedUsesChatCompletionsPath(t *testing.T) {
|
||||
|
||||
@@ -530,9 +530,16 @@ func (c *CodexQuotaOverdraftCoordinator) clearQuotaPause(accountID int64, state
|
||||
return
|
||||
}
|
||||
clearedSchedulingState := false
|
||||
if state != nil && sameCodexQuotaOverdraftTime(account.RateLimitResetAt, state.ObservedRateLimit) {
|
||||
_ = c.accountRepo.ClearRateLimit(ctx, accountID)
|
||||
clearedSchedulingState = true
|
||||
probeProvedAvailable := state != nil &&
|
||||
(state.Status == codexQuotaOverdraftProbePassed || state.Status == codexQuotaOverdraftProbeRecovered)
|
||||
if probeProvedAvailable && account.RateLimitResetAt != nil {
|
||||
staleResetAt := account.RateLimitResetAt.UTC()
|
||||
if err := c.accountRepo.ClearRateLimit(ctx, accountID); err != nil {
|
||||
slog.Warn("codex_quota_overdraft_stale_rate_limit_clear_failed", "account_id", accountID, "reset_at", staleResetAt, "error", err)
|
||||
} else {
|
||||
clearedSchedulingState = true
|
||||
slog.Info("codex_quota_overdraft_stale_rate_limit_cleared", "account_id", accountID, "reset_at", staleResetAt, "probe_status", state.Status)
|
||||
}
|
||||
}
|
||||
if IsAccountSchedulingThresholdReason(account.TempUnschedulableReason) || codexQuotaOverdraftPauseReason(account.TempUnschedulableReason) {
|
||||
_ = c.accountRepo.ClearTempUnschedulable(ctx, accountID)
|
||||
@@ -896,17 +903,6 @@ func codexQuotaOverdraftPauseReason(reason string) bool {
|
||||
return ok && payload.Source == codexQuotaOverdraftPauseSource
|
||||
}
|
||||
|
||||
func sameCodexQuotaOverdraftTime(left, right *time.Time) bool {
|
||||
if left == nil || right == nil {
|
||||
return false
|
||||
}
|
||||
delta := left.Sub(*right)
|
||||
if delta < 0 {
|
||||
delta = -delta
|
||||
}
|
||||
return delta <= time.Second
|
||||
}
|
||||
|
||||
func codexQuotaOverdraftResetAt(raw any, now time.Time) *time.Time {
|
||||
if raw == nil {
|
||||
return nil
|
||||
|
||||
@@ -22,6 +22,16 @@ type codexOverdraftProbeRepoStub struct {
|
||||
clearLimitCalls int
|
||||
}
|
||||
|
||||
type codexOverdraftRuntimeBlockerStub struct {
|
||||
clearCalls int
|
||||
}
|
||||
|
||||
func (b *codexOverdraftRuntimeBlockerStub) BlockAccountScheduling(*Account, time.Time, string) {}
|
||||
|
||||
func (b *codexOverdraftRuntimeBlockerStub) ClearAccountSchedulingBlock(int64) {
|
||||
b.clearCalls++
|
||||
}
|
||||
|
||||
func (r *codexOverdraftProbeRepoStub) GetByID(context.Context, int64) (*Account, error) {
|
||||
return r.account, nil
|
||||
}
|
||||
@@ -261,6 +271,49 @@ func TestCodexQuotaOverdraftRecoveryClearsWindowState(t *testing.T) {
|
||||
require.Zero(t, repo.tempPauseCalls)
|
||||
}
|
||||
|
||||
func TestCodexQuotaOverdraftAvailableStateClearsMismatchedStaleRateLimit(t *testing.T) {
|
||||
now := time.Date(2026, time.August, 14, 10, 0, 0, 0, time.UTC)
|
||||
for _, status := range []string{codexQuotaOverdraftProbePassed, codexQuotaOverdraftProbeRecovered} {
|
||||
t.Run(status, func(t *testing.T) {
|
||||
account := newCodexOverdraftProbeTestAccount(now)
|
||||
account.RateLimitedAt = codexQuotaOverdraftTimePtr(now.Add(-time.Minute))
|
||||
account.RateLimitResetAt = codexQuotaOverdraftTimePtr(now.Add(2 * time.Hour))
|
||||
repo := &codexOverdraftProbeRepoStub{account: account}
|
||||
blocker := &codexOverdraftRuntimeBlockerStub{}
|
||||
coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, runtimeBlocker: blocker}
|
||||
state := &CodexQuotaOverdraftProbeState{
|
||||
Status: status,
|
||||
ObservedRateLimit: codexQuotaOverdraftTimePtr(now.Add(time.Hour)),
|
||||
}
|
||||
|
||||
coordinator.clearQuotaPause(account.ID, state)
|
||||
|
||||
require.Equal(t, 1, repo.clearLimitCalls)
|
||||
require.Nil(t, account.RateLimitResetAt)
|
||||
require.Equal(t, 1, blocker.clearCalls)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCodexQuotaOverdraftFailedStateDoesNotClearRateLimit(t *testing.T) {
|
||||
now := time.Date(2026, time.August, 14, 10, 0, 0, 0, time.UTC)
|
||||
account := newCodexOverdraftProbeTestAccount(now)
|
||||
account.RateLimitResetAt = codexQuotaOverdraftTimePtr(now.Add(2 * time.Hour))
|
||||
repo := &codexOverdraftProbeRepoStub{account: account}
|
||||
blocker := &codexOverdraftRuntimeBlockerStub{}
|
||||
coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, runtimeBlocker: blocker}
|
||||
state := &CodexQuotaOverdraftProbeState{
|
||||
Status: codexQuotaOverdraftProbeFailed,
|
||||
ObservedRateLimit: codexQuotaOverdraftTimePtr(now.Add(time.Hour)),
|
||||
}
|
||||
|
||||
coordinator.clearQuotaPause(account.ID, state)
|
||||
|
||||
require.Zero(t, repo.clearLimitCalls)
|
||||
require.NotNil(t, account.RateLimitResetAt)
|
||||
require.Zero(t, blocker.clearCalls)
|
||||
}
|
||||
|
||||
func TestClassifyCodexQuotaOverdraftProbeResponses(t *testing.T) {
|
||||
status, reason := classifyCodexQuotaOverdraftProbe(http.StatusOK, nil, []byte(`data: {"type":"response.completed"}`))
|
||||
require.Equal(t, "available", status)
|
||||
|
||||
@@ -259,6 +259,7 @@ func ProvideAccountTestService(
|
||||
tlsFPProfileService *TLSFingerprintProfileService,
|
||||
openAIGatewayService *OpenAIGatewayService,
|
||||
settingService *SettingService,
|
||||
codexQuotaOverdraft *CodexQuotaOverdraftCoordinator,
|
||||
) *AccountTestService {
|
||||
service := NewAccountTestService(
|
||||
accountRepo,
|
||||
@@ -272,6 +273,7 @@ func ProvideAccountTestService(
|
||||
)
|
||||
service.agentIdentityWS = openAIGatewayService
|
||||
service.SetSettingService(settingService)
|
||||
service.SetCodexQuotaOverdraftCoordinator(codexQuotaOverdraft)
|
||||
return service
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user