diff --git a/CODEX_OVERDRAFT_DEPLOYMENT_CN.md b/CODEX_OVERDRAFT_DEPLOYMENT_CN.md index 1417c3e16..28b57e0a7 100644 --- a/CODEX_OVERDRAFT_DEPLOYMENT_CN.md +++ b/CODEX_OVERDRAFT_DEPLOYMENT_CN.md @@ -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` diff --git a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md index 5a2efaef7..ec5f417f6 100644 --- a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md +++ b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md @@ -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 ``` 数据库状态可这样检查： diff --git a/README_CN.md b/README_CN.md index ad52e4472..dbfe54d73 100644 --- a/README_CN.md +++ b/README_CN.md @@ -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% 时没有探测日志是正常现象。 ## 来源、许可证与风险 diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 83b97e68f..5f2d62856 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -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) diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index 9e1d29c2f..450af91c8 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -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{ { diff --git a/backend/internal/service/account_test_service_openai_compact_test.go b/backend/internal/service/account_test_service_openai_compact_test.go index 5b973285a..c5cb7b9f8 100644 --- a/backend/internal/service/account_test_service_openai_compact_test.go +++ b/backend/internal/service/account_test_service_openai_compact_test.go @@ -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"]) diff --git a/backend/internal/service/account_test_service_openai_test.go b/backend/internal/service/account_test_service_openai_test.go index 2cd63ebe3..00813e04e 100644 --- a/backend/internal/service/account_test_service_openai_test.go +++ b/backend/internal/service/account_test_service_openai_test.go @@ -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) { diff --git a/backend/internal/service/openai_codex_quota_overdraft_probe.go b/backend/internal/service/openai_codex_quota_overdraft_probe.go index cde930e0a..8ae318ca1 100644 --- a/backend/internal/service/openai_codex_quota_overdraft_probe.go +++ b/backend/internal/service/openai_codex_quota_overdraft_probe.go @@ -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 diff --git a/backend/internal/service/openai_codex_quota_overdraft_probe_test.go b/backend/internal/service/openai_codex_quota_overdraft_probe_test.go index fa1d08354..e1f2f68fc 100644 --- a/backend/internal/service/openai_codex_quota_overdraft_probe_test.go +++ b/backend/internal/service/openai_codex_quota_overdraft_probe_test.go @@ -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) diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index 2b58b8797..1e853b64c 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -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 }