diff --git a/.downstream/codex-quality-diagnostics.md b/.downstream/codex-quality-diagnostics.md new file mode 100644 index 000000000000..a6b0bb7e81cd --- /dev/null +++ b/.downstream/codex-quality-diagnostics.md @@ -0,0 +1,51 @@ +# Codex 固定账号质量诊断 + +此入口用于管理员在真实 `/v1/responses` 链路上运行有界对照。路由材料、完整响应、上游模型声明和回答质量是四种不同证据；采集成功不代表回答质量恢复。 + +## 接口与范围 + +所有管理操作位于 `/api/v1/admin/accounts/:id/codex-quality-runs`，要求真实管理员身份。运行绑定该管理员已有的业务 Key、其分组、目标 Pro OAuth 账号主体、代理、`gpt-6-astra`、显式 `high` 和题面 SHA-256。 + +| 操作 | 请求 | 行为 | +| --- | --- | --- | +| `POST` 基础路径 | `run_id`、`api_key_id`、`prompt_sha256`、`max_sends`、`ttl_seconds` | 创建或恢复同一运行，换发短时 grant；最多 60 次发送、最长 7200 秒 | +| `GET /:run_id` | 无正文 | 读取预算、原始上游观测和私有路由资格 | +| `POST /:run_id/renew-route` | `operation_id` | 显式采集与业务出口复验，均计入同一预算 | +| `POST /:run_id/close` | 空对象 | 关闭 grant，释放该运行自身连接，擦除私有 Cookie 值，保留账本 | + +`run_id`、`operation_id` 与 trial 均使用规范化 UUID。同一 run 恢复时不能改变绑定范围或增加预算，换发 grant 不恢复已花费次数；closed 运行不能重开。 + +业务仍用原 Key 作为 `Authorization: Bearer`，额外携带 `X-Sub2API-Quality-Grant` 和 `X-Sub2API-Quality-Trial`。正文的 `input` 必须为绑定的原题字符串，`model=gpt-6-astra`、`reasoning.effort=high`、`stream=true`、`store=false`。客户端会话元数据与 prompt cache key 可保留；不支持把此入口扩展为任意模型或任意账号转发。 + +grant 与 trial header 在 handler 开始时移入私有执行上下文，不转发给上游。错误、过期或已消费的授权直接失败，不回落到普通选号。 + +## 执行约束 + +- 只在私有选号判断中跳过账号手动停调度开关；账号、Key、分组和代理配置不修改。当前 Key 状态、到期、额度、渠道模型限制、账号健康、并发及路由资格仍检查。 +- acquire、verify、business 共用持久 CAS 预算，在本地准备与检查结束、上游 I/O 开始前预留。传输结果未知保留为 unknown，不能重放该 trial；预留次数是保守的发送上限，不能宣称每次预留均已得到上游响应。 +- 诊断材料使用独立 bundle、clock 与连接，不能覆盖普通资格，不登记自动续获。GET 不抢占或恢复普通 half-open probe。 +- 普通健康、调度恢复、proxy circuit 和自动恢复通知与诊断隔离；实际用量账单及额度快照按真实请求记录。 +- 记录来源是网关改写前的原始响应。`created_model`、`terminal_model`、HTTP/事件模型头分别保留；模型头冲突保存在 `header_models`。缺失 token 数据为 null，不替换为零。 +- `completed` 仅代表成功终态观测，不能替代模型核验或数学判分。当前仍保留修前探针形态，探针未显式给出的 effort 如实记录为空。 + +## 运维执行器 + +项目运维目录使用 `scripts/invoke-codex-quality-acceptance.py`。它从既有配置读取管理员凭据，仅在内存保存访问令牌、业务 Key 和 grant。固定账本为 `artifacts/evidence/codex-quality-20260923.json`；跨日、客户端中断、换版本都不能换 run ID 或清空预算。 + +```powershell +python scripts/invoke-codex-quality-acceptance.py preflight +python scripts/invoke-codex-quality-acceptance.py local-status +python scripts/invoke-codex-quality-acceptance.py status +``` + +`baseline` 与 `verification` 为真实生成操作，必须传发布者核实的 `--version` 和 `--source-sha`。前者 6 次，后者 12 次；前两次只观察，修后计分序列跨至少 6 分钟与两次实际资格续获。所有采集、复验、失败和对照都受同一 60 次上限约束。生成期间由单一执行者持有本地账本锁，发布者不得同时替换容器。 + +`review` 只对已有 trial 登记数学结论，不调用模型。通过条件包括完整响应、固定账号与最终 high、最终原始模型匹配、全部已提供模型头一致，以及答案和证明均正确；不能用正则找到“21”便自动通过。题目基准为 9 圆＋12 星，且需证明 20 不足。 + +`reconcile` 不发送模型请求。只有收到应用明确的发送前拒绝、服务器账本也无该 trial，才允许保留原记录并安排新 trial；网络超时或代理错误不能当作“没发送”。 + +## 发布次序 + +诊断基础设施先独立发布并采集修前基线。协议与账号身份修复随后发布，在共同宿主基线、同一账号和同一预算下完成修后序列。不能先应用修复再把结果命名为修前基线。 + +后续协议整合必须保持读取次序：网络 → 原始诊断观察 → 模型声明检查 → 业务响应改写。诊断业务流的读取限界需与普通业务一致，轻量采集探针保持自身预算。 diff --git a/.downstream/gpt6-routing-and-prompts.md b/.downstream/gpt6-routing-and-prompts.md index a0223928bd64..28c4327042f8 100644 --- a/.downstream/gpt6-routing-and-prompts.md +++ b/.downstream/gpt6-routing-and-prompts.md @@ -1,6 +1,6 @@ # GPT-6 模型、Codex 路由与提示词 -本轮以官方 v0.2.7 和下游生产提交 `300bd6dc5c40191d935cc08db910942f996bb26f` 为基线。宿主目标版本为 `0.2.7-codexrip.2`，七个内置插件为 `0.2.8`；新版插件要求宿主至少为该下游版本。宿主每次发布都会重新签名内置包，已安装的同版本号不能换包，因此 `0.2.7-codexrip.4` 起七个插件为 `0.2.9`，合并官方 v0.2.8 后的 `0.2.8-codexrip.1` 起为 `0.2.10`（宿主范围放宽到 `<0.2.9`，下限不变，便于回退到 0.2.7 宿主）。官方 v0.2.8 自带 GPT-6 Sol/Luna 定价与识别，下游改用官方实现，只保留 Codex 专属的路由门控、指令模板、实时容量和日期快照不改名的约定。插件固定版本意图继续保留，旧提示词插件不能解释新规则时明确报告不可用。 +本轮以官方 v0.2.7 和下游生产提交 `300bd6dc5c40191d935cc08db910942f996bb26f` 为基线。宿主目标版本为 `0.2.7-codexrip.2`，七个内置插件为 `0.2.8`；新版插件要求宿主至少为该下游版本。宿主每次发布都会重新签名内置包，已安装的同版本号不能换包，因此 `0.2.7-codexrip.4` 起七个插件为 `0.2.9`，合并官方 v0.2.8 后的 `0.2.8-codexrip.1` 起为 `0.2.10`，加入 Codex 质量诊断的 `0.2.8-codexrip.2` 起为 `0.2.11`（宿主范围放宽到 `<0.2.9`，下限不变，便于回退到 0.2.7 宿主）。官方 v0.2.8 自带 GPT-6 Sol/Luna 定价与识别，下游改用官方实现，只保留 Codex 专属的路由门控、指令模板、实时容量和日期快照不改名的约定。插件固定版本意图继续保留，旧提示词插件不能解释新规则时明确报告不可用。 ## 模型同步与容量 diff --git a/.github/scripts/test_first_party_plugins.sh b/.github/scripts/test_first_party_plugins.sh index 027c1df8ce95..1e8ba8b1934a 100644 --- a/.github/scripts/test_first_party_plugins.sh +++ b/.github/scripts/test_first_party_plugins.sh @@ -22,7 +22,7 @@ go build -trimpath -o "$plugin_build_dir/package-plugin" ./cmd/package-plugin -bundle-source ../plugins/bundle.source.json \ -binary-dir "$plugin_build_dir/binaries" \ -platform linux-amd64 \ - -tested-host-version 0.2.8-codexrip.1 \ + -tested-host-version 0.2.8-codexrip.2 \ -output "$plugin_build_dir/bundle" \ -signing-key-file "$plugin_build_dir/publisher.key" \ -key-id codexrip-plugins-test diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml index 6d4b31a3167e..7b097256dd3f 100644 --- a/.github/workflows/downstream-verify.yml +++ b/.github/workflows/downstream-verify.yml @@ -54,7 +54,7 @@ jobs: run: bash .github/scripts/test_model_context_capacity.sh - name: Service tests working-directory: backend - run: go test -p=1 -tags unit ./internal/service -run 'Test.*(Refusal|CyberPolicy|Continuation|InvalidEncryptedContent|PreviousResponseNotFound|CapacityShed|RequestScopedTransient|ServerOverloaded|BusinessSystemPrompt|RemoteSkill|Cindy|CompatibilityModes|AlphaSearch)|Test(UpdateService|NormalizeAccountTaxonomyName|FilterConsoleAccounts|AccountConsolePlan|AccountFacetMatcher)' -count=1 + run: go test -p=1 -tags unit ./internal/service -run 'Test.*(Refusal|CyberPolicy|Continuation|InvalidEncryptedContent|PreviousResponseNotFound|CapacityShed|RequestScopedTransient|ServerOverloaded|BusinessSystemPrompt|RemoteSkill|Cindy|CompatibilityModes|AlphaSearch|CodexQuality)|Test(UpdateService|NormalizeAccountTaxonomyName|FilterConsoleAccounts|AccountConsolePlan|AccountFacetMatcher)' -count=1 - name: Upstream automation classifier tests run: | python3 .github/scripts/test_upstream_risk.py diff --git a/backend/internal/handler/admin/account_codex_quality.go b/backend/internal/handler/admin/account_codex_quality.go new file mode 100644 index 000000000000..4711acb3ee8e --- /dev/null +++ b/backend/internal/handler/admin/account_codex_quality.go @@ -0,0 +1,113 @@ +package admin + +import ( + "context" + "strconv" + + "github.com/Wei-Shaw/sub2api/internal/pkg/response" + "github.com/Wei-Shaw/sub2api/internal/server/middleware" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" +) + +func codexQualityAdmin(c *gin.Context) (int64, int64, bool) { + c.Header("Cache-Control", "no-store") + actor, ok := middleware.GetAuthSubjectFromContext(c) + role, roleOK := middleware.GetUserRoleFromContext(c) + id, err := strconv.ParseInt(c.Param("id"), 10, 64) + if !ok || !roleOK || role != "admin" || actor.UserID <= 0 || err != nil || id <= 0 { + response.Error(c, 403, "Codex quality run unavailable") + return 0, 0, false + } + return actor.UserID, id, true +} + +func writeCodexQualityResult(c *gin.Context, result *service.CodexQualityRunView, err error) { + if err != nil || result == nil { + response.Error(c, 409, "Codex quality run unavailable") + return + } + response.Success(c, result) +} + +func (h *AccountHandler) CreateCodexQualityRun(c *gin.Context) { + actor, id, ok := codexQualityAdmin(c) + if !ok { + return + } + var req service.CodexQualityCreateRequest + if h.codexTicketGateway == nil || c.ShouldBindJSON(&req) != nil || req.APIKeyID <= 0 { + response.Error(c, 409, "Codex quality run unavailable") + return + } + // A global integration admin key resolves to a real administrator too. Only + // that administrator's own existing downstream key can authorize this run. + keys, _, err := h.adminService.GetUserAPIKeys(c.Request.Context(), actor, 1, 1000, "", "") + if err != nil { + response.Error(c, 409, "Codex quality run unavailable") + return + } + var key *service.APIKey + for index := range keys { + if keys[index].ID == req.APIKeyID && keys[index].UserID == actor { + key = &keys[index] + break + } + } + result, err := h.codexTicketGateway.CreateCodexQualityRun(c.Request.Context(), actor, id, key, req) + writeCodexQualityResult(c, result, err) +} + +func (h *AccountHandler) ReadCodexQualityRun(c *gin.Context) { + actor, id, ok := codexQualityAdmin(c) + if !ok { + return + } + if h.codexTicketGateway == nil { + response.Error(c, 409, "Codex quality run unavailable") + return + } + result, err := h.codexTicketGateway.ReadCodexQualityRun(c.Request.Context(), actor, id, c.Param("run_id")) + writeCodexQualityResult(c, result, err) +} + +func (h *AccountHandler) RenewCodexQualityRoute(c *gin.Context) { + actor, id, ok := codexQualityAdmin(c) + if !ok { + return + } + var req struct { + OperationID string `json:"operation_id"` + } + if h.codexTicketGateway == nil || c.ShouldBindJSON(&req) != nil { + response.Error(c, 409, "Codex quality run unavailable") + return + } + lookup := func(ctx context.Context, keyID int64) (*service.APIKey, error) { + keys, _, err := h.adminService.GetUserAPIKeys(ctx, actor, 1, 1000, "", "") + if err != nil { + return nil, service.ErrCodexQualityUnavailable + } + for index := range keys { + if keys[index].ID == keyID && keys[index].UserID == actor { + return &keys[index], nil + } + } + return nil, service.ErrCodexQualityUnavailable + } + result, err := h.codexTicketGateway.RenewCodexQualityRoute(c.Request.Context(), actor, id, c.Param("run_id"), req.OperationID, lookup) + writeCodexQualityResult(c, result, err) +} + +func (h *AccountHandler) CloseCodexQualityRun(c *gin.Context) { + actor, id, ok := codexQualityAdmin(c) + if !ok { + return + } + if h.codexTicketGateway == nil { + response.Error(c, 409, "Codex quality run unavailable") + return + } + result, err := h.codexTicketGateway.CloseCodexQualityRun(c.Request.Context(), actor, id, c.Param("run_id")) + writeCodexQualityResult(c, result, err) +} diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index cd7fb40b3d7a..668c51b17f9c 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -485,6 +485,7 @@ func NewOpenAIGatewayHandler( // Responses handles OpenAI Responses API endpoint // POST /openai/v1/responses func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { + service.StageCodexQualityHeaders(c) // 局部兜底：确保该 handler 内部任何 panic 都不会击穿到进程级。 streamStarted := false defer h.recoverResponsesPanic(c, &streamStarted) @@ -546,6 +547,10 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "Request body is empty") return } + if err := h.gatewayService.AdmitCodexQualityRequest(c, apiKey, body, h.apiKeyService.GetByID); err != nil { + h.errorResponse(c, http.StatusConflict, "codex_quality_unavailable", "Codex quality run unavailable") + return + } setOpsRequestContext(c, "", false) sessionHashBody := body @@ -923,7 +928,10 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { scheduleDecision service.OpenAIAccountScheduleDecision err error ) - if retryingSameAccount { + if service.IsCodexQualityRequest(c.Request.Context()) { + selection, err = h.gatewayService.SelectCodexQualityAccount(c.Request.Context()) + scheduleDecision.Layer = "codex_quality" + } else if retryingSameAccount { selection, err = h.gatewayService.ReacquireOpenAISameAccountSelection(c.Request.Context(), sameAccountRetrySelection) sameAccountRetrySelection = nil scheduleDecision.Layer = "same_account_retry" @@ -1037,6 +1045,10 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { accountReleaseFunc, slotResult = h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, reqStream, &streamStarted, reqLog) } if slotResult == openAISlotAcquireProfitVetoed { + if service.IsCodexQualityRequest(c.Request.Context()) { + h.handleStreamingAwareError(c, http.StatusConflict, "codex_quality_unavailable", "Codex quality run unavailable", streamStarted) + return + } // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错； // 否决次数达上限则直接终止，避免排队抢槽后才终检的延迟放大。 if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { @@ -1121,6 +1133,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { stampOpenAIRequestedReasoningEffort(res, c) usageSnapshot := snapshotOpenAIUsageMetadata(c, apiKey, account, subscription, channelMapping, reqModel, res, body) usageInput := usageSnapshot.Input(res, h.apiKeyService, pricingAt) + usageInput.CodexQuality = service.IsCodexQualityRequest(c.Request.Context()) usageAPIKeyID := usageInput.APIKey.ID usageGroupID := usageInput.APIKey.GroupID usageAccountID := usageInput.Account.ID @@ -1137,6 +1150,14 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { } }) } + if err != nil && service.IsCodexQualityRequest(c.Request.Context()) { + h.gatewayService.ReleaseOpenAIRuntimeBreakerProbeForSelection(selection) + if shouldSubmitOpenAIUsage(err, result) { + submitResponsesUsage(result) + } + h.handleStreamingAwareError(c, http.StatusBadGateway, "codex_quality_failed", "Diagnostic attempt did not complete", streamStarted || c.Writer.Written()) + return + } if err != nil { if result != nil && result.ClientDisconnect { reqLog.Info("openai.client_disconnected", @@ -2680,7 +2701,7 @@ func (h *OpenAIGatewayHandler) acquireResponsesAccountSlotWithFailureResponse( selection.Account = latest // 调度器已抢槽路径无门时由选号内部完成 eager 绑定；门下选号内部 // 推迟绑定，这里在终检通过后补准入后绑定。 - if selection.ProfitGateActive() { + if selection.ProfitGateActive() && !service.IsCodexQualityRequest(ctx) { if err := h.gatewayService.BindStickySessionAfterProfitAdmission(ctx, groupID, sessionHash, account.ID); err != nil { reqLog.Warn("openai.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) } diff --git a/backend/internal/handler/openai_official_http_failover.go b/backend/internal/handler/openai_official_http_failover.go index fff701e55afd..3d9c36c8e345 100644 --- a/backend/internal/handler/openai_official_http_failover.go +++ b/backend/internal/handler/openai_official_http_failover.go @@ -40,6 +40,10 @@ func (h *OpenAIGatewayHandler) reportOpenAIHTTPAccountScheduleResult( firstTokenMs *int, observedErr error, ) { + if c != nil && c.Request != nil && service.IsCodexQualityRequest(c.Request.Context()) { + h.gatewayService.ReleaseOpenAIRuntimeBreakerProbeForSelection(selection) + return + } if isOpenAIOfficialHTTPFailover(c, account) { if !success && failoverClientGone(c) { return diff --git a/backend/internal/repository/codex_quality_breaker.go b/backend/internal/repository/codex_quality_breaker.go new file mode 100644 index 000000000000..545081652a1a --- /dev/null +++ b/backend/internal/repository/codex_quality_breaker.go @@ -0,0 +1,18 @@ +package repository + +import "context" + +// A retained marker/claim is an ordinary half-open breaker. Diagnostics wait +// for its ordinary owner to recover it, rather than claiming or deleting it. +func (c *gatewayCache) CodexQualityRuntimeBlocked(ctx context.Context, accountID int64, models []string) (bool, error) { + var keys []string + for _, model := range normalizeOpenAIRuntimeBreakerModels(models) { + base := openAIRuntimeBreakerBaseKey(accountID, model) + keys = append(keys, base+":block", base+":marker", base+":claim") + } + if len(keys) == 0 { + return false, nil + } + count, err := c.rdb.Exists(ctx, keys...).Result() + return count > 0, err +} diff --git a/backend/internal/repository/codex_quality_lease.go b/backend/internal/repository/codex_quality_lease.go new file mode 100644 index 000000000000..341bcc5f11e3 --- /dev/null +++ b/backend/internal/repository/codex_quality_lease.go @@ -0,0 +1,21 @@ +package repository + +import "time" + +func (s *httpUpstreamService) HasCodexQualityConnection(id string, accountID int64, scope string) bool { + s.mu.RLock() + defer s.mu.RUnlock() + lease := s.codexConnectionLeases[id] + return lease != nil && lease.accountID == accountID && lease.scope == scope && time.Now().Before(lease.expiresAt) +} + +// Close only the connection whose complete ownership tuple belongs to this +// diagnostic run. Shared pools and other accounts' leases are never touched. +func (s *httpUpstreamService) CloseCodexQualityConnection(id string, accountID int64, scope string) { + s.mu.RLock() + lease := s.codexConnectionLeases[id] + s.mu.RUnlock() + if lease != nil && lease.accountID == accountID && lease.scope == scope { + s.removeCodexConnectionLease(id, lease) + } +} diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index 0705b15f646b..c1fc86e56b60 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -532,6 +532,12 @@ func registerAccountRoutes(admin *gin.RouterGroup, h *handler.Handlers, stepUpAu accounts.POST("/:id/codex-tickets/stop", h.Admin.Account.StopCodexTicketRenewal) accounts.GET("/:id/codex-fingerprint", h.Admin.Plugin.RegisterResource(extensionv1.ResourceDescriptor{ResourceGrant: extensionv1.ResourceGrant{Name: "codex.fingerprint", Capability: extensionv1.CapabilityRequest, Permission: "admin"}, OwnerPluginKey: "codexrip.codex-runtime", AccountParam: "id", Method: "GET", Path: accounts.BasePath() + "/:id/codex-fingerprint"}), h.Admin.Account.CodexFingerprint) accounts.POST("/:id/codex-routing/validate", h.Admin.Plugin.RegisterResource(extensionv1.ResourceDescriptor{ResourceGrant: extensionv1.ResourceGrant{Name: "codex.routing.validate", Capability: extensionv1.CapabilityCredentials, Permission: "admin"}, OwnerPluginKey: "codexrip.codex-runtime", AccountParam: "id", Method: "POST", Path: accounts.BasePath() + "/:id/codex-routing/validate"}), h.Admin.Account.ValidateCodexRouting) + // Host-private, administrator-owned diagnostics. Each operation rechecks + // the existing credentials binding through the named routing host broker. + accounts.POST("/:id/codex-quality-runs", h.Admin.Account.CreateCodexQualityRun) + accounts.GET("/:id/codex-quality-runs/:run_id", h.Admin.Account.ReadCodexQualityRun) + accounts.POST("/:id/codex-quality-runs/:run_id/renew-route", h.Admin.Account.RenewCodexQualityRoute) + accounts.POST("/:id/codex-quality-runs/:run_id/close", h.Admin.Account.CloseCodexQualityRun) accounts.PUT("/:id/codex-fingerprint/profile", h.Admin.Plugin.RegisterResource(extensionv1.ResourceDescriptor{ResourceGrant: extensionv1.ResourceGrant{Name: "codex.fingerprint.profile", Capability: extensionv1.CapabilityRequest, Permission: "admin"}, RequiredCapabilities: []string{extensionv1.CapabilityAdmin}, OwnerPluginKey: "codexrip.codex-runtime", AccountParam: "id", Method: "PUT", Path: accounts.BasePath() + "/:id/codex-fingerprint/profile"}), h.Admin.Account.SelectCodexProfile) accounts.POST("/:id/recover-state", h.Admin.Account.RecoverState) accounts.POST("/:id/cindy-balance/recover", h.Admin.Plugin.RegisterResource(extensionv1.ResourceDescriptor{ResourceGrant: extensionv1.ResourceGrant{Name: "cindy.balance.recover", Capability: extensionv1.CapabilityProvider, Permission: "admin"}, RequiredCapabilities: []string{extensionv1.CapabilityAdmin}, OwnerPluginKey: service.CindyAccountViewPluginKey, AccountParam: "id", Method: "POST", Path: accounts.BasePath() + "/:id/cindy-balance/recover"}), h.Admin.Account.ClearCindyBalanceInsufficient) diff --git a/backend/internal/service/codex_quality_health.go b/backend/internal/service/codex_quality_health.go new file mode 100644 index 000000000000..21c1f4a06246 --- /dev/null +++ b/backend/internal/service/codex_quality_health.go @@ -0,0 +1,58 @@ +package service + +import ( + "context" + "time" +) + +// Diagnostics never clear an ordinary breaker or borrow its half-open probe. +// The production adapter implements this with a read-only Redis command. +type codexQualityBreakerReader interface { + CodexQualityRuntimeBlocked(context.Context, int64, []string) (bool, error) +} + +func (s *OpenAIGatewayService) codexQualityHealthBlocked(ctx context.Context, account *Account) bool { + now := time.Now() + if value, exists := s.openaiAccountRuntimeBlockUntil.Load(account.ID); exists { + until, valid := value.(time.Time) + if !valid || until.IsZero() || now.Before(until) { + return true + } + } + model := openAIAccountModelTransientModel(codexQualityModel) + state := s.getOpenAIAccountModelTransientState() + state.mu.Lock() + entry := state.entries[openAIAccountModelKey{AccountID: account.ID, Model: model}] + modelBlocked := now.Before(entry.blockUntil) + state.mu.Unlock() + if modelBlocked { + return true + } + if proxyID, exists := openAIProxyStreamCircuitProxyID(account); exists { + circuit := s.getOpenAIProxyStreamCircuit() + circuit.mu.Lock() + blocked := !circuit.settings.disabled && now.Before(circuit.entries[proxyID].blockedUntil) + circuit.mu.Unlock() + if blocked { + return true + } + } + if s.cache != nil { + reader, ok := s.cache.(codexQualityBreakerReader) + if !ok { + return true + } + blocked, err := reader.CodexQualityRuntimeBlocked(ctx, account.ID, []string{"", model}) + if err != nil || blocked { + return true + } + } + if s.rateLimitService != nil && s.rateLimitService.settingService != nil { + thresholds := s.rateLimitService.settingService.GetAccountSchedulingThresholds(ctx) + decision := EvaluateAccountSchedulingThreshold(account, thresholds, now) + if decision.ShouldPause && decision.Until != nil && now.Before(*decision.Until) { + return true + } + } + return false +} diff --git a/backend/internal/service/codex_quality_manual_stop_test.go b/backend/internal/service/codex_quality_manual_stop_test.go new file mode 100644 index 000000000000..ed4962e2b2e5 --- /dev/null +++ b/backend/internal/service/codex_quality_manual_stop_test.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "net/http" + "strings" + "testing" + + extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +type qualityManualStopDirectory struct { + *routingHostDirectoryFixture + scopeReads int +} + +func (d *qualityManualStopDirectory) PrepareCodexRoutingScope(context.Context, int64, string) (extensionv1.CodexRoutingScope, error) { + d.scopeReads++ + return d.scope, nil +} + +type qualityManualStopConcurrency struct { + ConcurrencyCache + acquisitions int +} + +func (c *qualityManualStopConcurrency) AcquireAccountSlot(context.Context, int64, int, string) (bool, error) { + c.acquisitions++ + return true, nil +} + +func (*qualityManualStopConcurrency) ReleaseAccountSlot(context.Context, int64, string) error { + return nil +} + +func TestCodexQualityManualStopRequiredBeforeCreateRenewAndSend(t *testing.T) { + ctx := context.Background() + host, baseDirectory, store := routingHostFixture() + installation := host.installation + installation.ID, installation.State, installation.Version = 1, PluginStateEnabled, "0.2.7" + installation.Manifest.Requires.ExtensionAPI = 1 + installation.Bindings = []PluginBinding{{Capability: extensionv1.CapabilityCredentials, Platform: PlatformOpenAI, AccountType: AccountTypeOAuth, Enabled: true, RolloutPercent: 100}} + store.PluginRepository = &pluginTokenRepository{installation: installation} + directory := &qualityManualStopDirectory{routingHostDirectoryFixture: baseDirectory} + manager := NewPluginManager(store, nil, nil, PluginHostInfo{}, newFakePluginKVStore()) + manager.accountDirectory = directory + manager.extensions.Store(&pluginExtensionRegistry{installations: map[int64]*PluginInstallation{1: installation}, runtimes: map[int64]*pluginRuntime{}}) + group, proxy := int64(53), int64(34) + account := &Account{ID: 16380, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: false, Concurrency: 1, GroupIDs: []int64{group}, ProxyID: &proxy, Credentials: map[string]any{"plan_type": "pro", "chatgpt_account_id": "quality-account"}} + directory.account = *extensionAccount(account) + directory.scope.AccountID, directory.scope.Identity = account.ID, CodexTicketAccountIdentity(account) + slots := &qualityManualStopConcurrency{} + s := &OpenAIGatewayService{pluginManager: manager, accountRepo: &routingAccountRepositoryFixture{account: account}, concurrencyService: &ConcurrencyService{cache: slots}} + key := &APIKey{ID: 101, UserID: 9, GroupID: &group, Status: StatusAPIKeyActive} + lookup := func(context.Context, int64) (*APIKey, error) { return key, nil } + request := CodexQualityCreateRequest{RunID: uuid.NewString(), APIKeyID: key.ID, PromptSHA256: codexQualityHash("question"), MaxSends: 6, TTLSeconds: 7200} + created, err := s.CreateCodexQualityRun(ctx, key.UserID, account.ID, key, request) + require.NoError(t, err, "the stopped account remains eligible for its bounded diagnostic run") + require.NotEmpty(t, created.Grant) + require.False(t, account.Schedulable, "the eligibility copy must not enable the persisted account") + run, revision, err := readCodexQualityRun(ctx, store, request.RunID) + require.NoError(t, err) + + // Simulate an administrator turning ordinary scheduling back on while the + // existing diagnostic run and its downstream grant remain unexpired. + account.Schedulable = true + otherRequest := request + otherRequest.RunID = uuid.NewString() + _, err = s.CreateCodexQualityRun(ctx, key.UserID, account.ID, key, otherRequest) + require.ErrorIs(t, err, ErrCodexQualityUnavailable) + missing, missingRevision, err := readCodexQualityRun(ctx, store, otherRequest.RunID) + require.NoError(t, err) + require.Empty(t, missing.RunID) + require.Zero(t, missingRevision, "a scheduled account must not obtain a new run or grant") + + scopeReads := directory.scopeReads + _, err = s.RenewCodexQualityRoute(ctx, key.UserID, account.ID, request.RunID, uuid.NewString(), lookup) + require.ErrorIs(t, err, ErrCodexQualityUnavailable) + require.Equal(t, scopeReads, directory.scopeReads, "reject before route probing, not at a later missing dependency") + require.Zero(t, slots.acquisitions) + require.Zero(t, directory.requests) + + runtime, err := s.codexQualityRuntime() + require.NoError(t, err) + for _, stage := range []string{"acquire", "verify", "business"} { + t.Run(stage, func(t *testing.T) { + execution := &codexQualityExecution{runtime: runtime, runID: run.RunID, grantDigest: run.GrantDigest, accountID: account.ID, stage: stage, trialID: uuid.NewString(), operationID: uuid.NewString(), keyLookup: lookup} + sendCtx := context.WithValue(ctx, codexQualityExecutionKey{}, execution) + req, err := http.NewRequestWithContext(sendCtx, http.MethodPost, "https://chatgpt.com/backend-api/codex/responses", strings.NewReader(`{"model":"gpt-6-astra","reasoning":{"effort":"high"}}`)) + require.NoError(t, err) + require.ErrorIs(t, reserveCodexQualitySend(req, account, nil), ErrCodexQualityUnavailable) + require.Equal(t, scopeReads, directory.scopeReads, "the fresh manual-stop guard must reject before qualification checks") + saved, savedRevision, err := readCodexQualityRun(ctx, store, run.RunID) + require.NoError(t, err) + require.Equal(t, revision, savedRevision) + require.Zero(t, saved.UsedSends) + require.Empty(t, saved.Attempts) + require.Zero(t, directory.requests) + require.True(t, account.Schedulable, "rejection must not flip the administrator's setting back off") + }) + } +} diff --git a/backend/internal/service/codex_quality_observation.go b/backend/internal/service/codex_quality_observation.go new file mode 100644 index 000000000000..335a258da007 --- /dev/null +++ b/backend/internal/service/codex_quality_observation.go @@ -0,0 +1,316 @@ +package service + +import ( + "bytes" + "context" + "io" + "net/http" + "slices" + "strings" + "sync" + "time" + + extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" + "github.com/klauspost/compress/zstd" + "github.com/tidwall/gjson" +) + +func codexQualityWireFields(req *http.Request) (string, string, error) { + if req == nil || req.GetBody == nil { + return "", "", ErrCodexQualityUnavailable + } + body, err := req.GetBody() + if err != nil { + return "", "", ErrCodexQualityUnavailable + } + defer func() { _ = body.Close() }() + var reader io.Reader = body + if strings.EqualFold(req.Header.Get("Content-Encoding"), "zstd") { + decoder, err := zstd.NewReader(body, zstd.WithDecoderMaxMemory(8<<20)) + if err != nil { + return "", "", ErrCodexQualityUnavailable + } + defer decoder.Close() + reader = decoder + } + raw, err := io.ReadAll(io.LimitReader(reader, (1<<20)+1)) + if err != nil || len(raw) > 1<<20 || !gjson.ValidBytes(raw) { + return "", "", ErrCodexQualityUnavailable + } + return gjson.GetBytes(raw, "model").String(), gjson.GetBytes(raw, "reasoning.effort").String(), nil +} + +func reserveCodexQualityAcquisition(req *http.Request, accountID int64) error { + e := codexQualityExecutionFromContext(req.Context()) + if e == nil { + return nil + } + if e.accountID != accountID || e.stage != "acquire" { + return ErrCodexQualityUnavailable + } + run, err := e.current(req.Context()) + if err != nil { + return err + } + account, err := e.runtime.account(req.Context(), run) + if err != nil { + return err + } + return reserveCodexQualitySend(req, account, nil) +} + +// Called after local request preparation, immediately before model IO. Every +// generation path of a quality run (acquire, verify and business) spends here. +func reserveCodexQualitySend(req *http.Request, account *Account, q *extensionv1.CodexRoutingQualification) error { + e := codexQualityExecutionFromContext(req.Context()) + if e == nil { + return nil + } + if account == nil || account.ID != e.accountID { + return ErrCodexQualityUnavailable + } + run, err := e.current(req.Context()) + if err != nil { + return err + } + if _, err = e.runtime.account(req.Context(), run); err != nil { + return err + } + model, effort, err := codexQualityWireFields(req) + if err != nil || model != codexQualityModel || (e.stage == "business" && effort != codexQualityEffort) { + return ErrCodexQualityUnavailable + } + if e.stage == "business" { + current, err := e.runtime.qualification(req.Context(), run) + if err != nil || q == nil || current.Bundle.Key != q.Bundle.Key || current.Bundle.Revision != q.Bundle.Revision { + return ErrCodexQualityUnavailable + } + } + e.requestModel, e.effort, e.qualification = model, effort, q + return reserveCodexQualityAttempt(e.runtime.ctx(req.Context()), e.runtime.store, e.runID, e.grantDigest, e.attempt()) +} + +func (e *codexQualityExecution) attempt() CodexQualityAttempt { + a := CodexQualityAttempt{Stage: e.stage, TrialID: e.trialID, OperationID: e.operationID, AccountID: e.accountID, RequestModel: e.requestModel, ReasoningEffort: e.effort, ResponseModels: []string{}, HeaderModels: []string{}} + if q := e.qualification; q != nil { + if q.Scope.ConnectionLeaseID != "" { + a.ConnectionFingerprint = codexQualityHash(q.Scope.ConnectionLeaseID)[:16] + } + if q.Bundle.Key != "" { + a.QualificationFingerprint = codexQualityHash(q.Bundle.Key)[:16] + } + } + return a +} + +func observeCodexQualityResponse(ctx context.Context, response *http.Response, err error, q *extensionv1.CodexRoutingQualification) { + e := codexQualityExecutionFromContext(ctx) + if e == nil { + return + } + if q != nil { + e.qualification = q + } + a := e.attempt() + if err != nil || response == nil { + a.State, a.ErrorCode = "unknown", "transport" + finishCodexQualityAttempt(e.runtime.ctx(ctx), e.runtime.store, e.runID, a) + return + } + a.HTTPStatus = response.StatusCode + observedAt := time.Now().UTC() + deleted := false + if e.stage == "business" && q != nil { + deleted = e.runtime.s.recordCodexRoutingDeletions(ctx, e.runtime.installation, q, response.Header, observedAt) + } + for _, name := range []string{"openai-model", "x-openai-model"} { + for _, value := range response.Header.Values(name) { + recordCodexQualityHeaderModel(&a, value) + } + } + observer := &codexQualityObservedBody{ReadCloser: response.Body, sse: strings.Contains(strings.ToLower(response.Header.Get("Content-Type")), "text/event-stream"), sniff: response.Header.Get("Content-Type") == "", attempt: a} + observer.finish = func(a CodexQualityAttempt) { + finishCodexQualityAttempt(e.runtime.ctx(ctx), e.runtime.store, e.runID, a) + if e.stage == "business" { + if _, err := e.current(ctx); err != nil { + return + } + matched := a.Completed && a.TerminalModel != nil && *a.TerminalModel == codexQualityModel && !deleted + var replacement *extensionv1.CodexRoutingQualification + if matched { + replacement = e.runtime.s.refreshObservedCodexCookies(ctx, e.runtime.installation, q, response.Header, extensionv1.CodexRoutingObservation{Stage: "business", Code: "routing_verified", RequestedModel: codexQualityModel, ResponseModel: codexQualityModel, Completed: true, ModelMatched: true, ObservedAt: observedAt}) + } + finish, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second) + defer cancel() + _, _ = mutateCodexQualityRun(e.runtime.ctx(finish), e.runtime.store, e.runID, func(run *codexQualityRun) error { + if run.Qualification != nil && e.qualification != nil && run.Qualification.Bundle == e.qualification.Bundle { + if !matched { + run.Qualification = nil + } else if replacement != nil { + run.Qualification = replacement + } + } + return nil + }) + } + } + response.Body = observer +} + +func qualitySafeModel(model string) string { + model = strings.TrimSpace(model) + if model == "" { + return "" + } + if len(model) > 128 || !strings.HasPrefix(model, "gpt-") { + return "other" + } + for _, r := range model { + if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._-", r) { + continue + } + return "other" + } + return model +} + +func recordCodexQualityHeaderModel(attempt *CodexQualityAttempt, value string) { + model := qualitySafeModel(value) + if model == "" { + return + } + if !slices.Contains(attempt.HeaderModels, model) && len(attempt.HeaderModels) < 8 { + attempt.HeaderModels = append(attempt.HeaderModels, model) + } + if len(attempt.HeaderModels) == 1 { + attempt.HeaderModel = &model + } else { + conflict := "conflicting_models" + attempt.HeaderModel = &conflict + } +} + +// A bounded single-frame observer; it never buffers the whole response and +// always forwards the original bytes without editing model fields or content. +type codexQualityObservedBody struct { + io.ReadCloser + sse, sniff, failed, oversized bool + pending []byte + attempt CodexQualityAttempt + once sync.Once + finish func(CodexQualityAttempt) +} + +func (b *codexQualityObservedBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + if n > 0 { + b.feed(p[:n]) + } + if err != nil { + b.complete(err) + } + return n, err +} + +func (b *codexQualityObservedBody) feed(raw []byte) { + if b.oversized { + return + } + b.pending = append(b.pending, raw...) + if b.sniff && (bytes.HasPrefix(bytes.TrimSpace(b.pending), []byte("data:")) || bytes.HasPrefix(bytes.TrimSpace(b.pending), []byte("event:"))) { + b.sse, b.sniff = true, false + } + if b.sse { + b.pending = bytes.ReplaceAll(b.pending, []byte("\r\n"), []byte("\n")) + for { + end := bytes.Index(b.pending, []byte("\n\n")) + if end < 0 { + break + } + var data []string + for _, line := range strings.Split(string(b.pending[:end]), "\n") { + if strings.HasPrefix(line, "data:") { + data = append(data, strings.TrimSpace(strings.TrimPrefix(line, "data:"))) + } + } + b.event([]byte(strings.Join(data, "\n"))) + b.pending = b.pending[end+2:] + } + } + if len(b.pending) > codexRoutingProbeReadLimit { + b.oversized, b.failed, b.pending = true, true, nil + } +} + +func (b *codexQualityObservedBody) event(raw []byte) { + if !gjson.ValidBytes(raw) { + return + } + kind := gjson.GetBytes(raw, "type").String() + root := gjson.ParseBytes(raw) + for _, path := range []string{"response.headers.openai-model", "response.headers.x-openai-model", "response.metadata.headers.openai-model", "response.metadata.headers.x-openai-model", "headers.openai-model", "headers.x-openai-model", "metadata.headers.openai-model", "metadata.headers.x-openai-model"} { + value := root.Get(path) + if value.Type == gjson.String { + recordCodexQualityHeaderModel(&b.attempt, value.String()) + } else if value.IsArray() { + for _, item := range value.Array() { + if item.Type == gjson.String { + recordCodexQualityHeaderModel(&b.attempt, item.String()) + } + } + } + } + r := root.Get("response") + if !r.Exists() { + r = root + } + model := qualitySafeModel(r.Get("model").String()) + if model != "" && !slices.Contains(b.attempt.ResponseModels, model) && len(b.attempt.ResponseModels) < 8 { + b.attempt.ResponseModels = append(b.attempt.ResponseModels, model) + } + if kind == "response.created" && model != "" { + b.attempt.CreatedModel = &model + } + if kind == "error" || kind == "response.failed" || kind == "response.incomplete" || (root.Get("error").Exists() && root.Get("error").Type != gjson.Null) { + b.failed = true + b.attempt.ErrorCode = "upstream_error" + } + if kind == "response.completed" || (root.Get("object").String() == "response" && r.Get("status").String() == "completed") { + if model != "" { + b.attempt.TerminalModel = &model + } + b.attempt.Completed = r.Get("status").String() == "completed" && (!r.Get("error").Exists() || r.Get("error").Type == gjson.Null) + optional := func(path string) *int64 { + value := r.Get(path) + if !value.Exists() || value.Type != gjson.Number || value.Int() < 0 { + return nil + } + n := value.Int() + return &n + } + b.attempt.ReasoningTokens = optional("usage.output_tokens_details.reasoning_tokens") + b.attempt.InputTokens, b.attempt.OutputTokens = optional("usage.input_tokens"), optional("usage.output_tokens") + } +} + +func (b *codexQualityObservedBody) complete(err error) { + b.once.Do(func() { + if !b.sse && !b.oversized { + b.event(b.pending) + } + b.attempt.Completed = b.attempt.Completed && !b.failed && b.attempt.HTTPStatus == http.StatusOK + b.attempt.State = "unknown" + if b.attempt.Completed { + b.attempt.State = "complete" + } else if b.failed || b.attempt.HTTPStatus >= 400 { + b.attempt.State, b.attempt.ErrorCode = "error", "upstream_error" + } else if err != nil && err != io.EOF { + b.attempt.ErrorCode = "stream_error" + } else { + b.attempt.ErrorCode = "incomplete" + } + b.finish(b.attempt) + }) +} + +func (b *codexQualityObservedBody) Close() error { b.complete(nil); return b.ReadCloser.Close() } diff --git a/backend/internal/service/codex_quality_route.go b/backend/internal/service/codex_quality_route.go new file mode 100644 index 000000000000..7e715642c516 --- /dev/null +++ b/backend/internal/service/codex_quality_route.go @@ -0,0 +1,105 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "time" + + extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" + "github.com/google/uuid" +) + +func (s *OpenAIGatewayService) RenewCodexQualityRoute(ctx context.Context, actor, accountID int64, id, operation string, lookup CodexQualityKeyLookup) (*CodexQualityRunView, error) { + canonical, valid := canonicalCodexQualityID(operation) + if !valid || lookup == nil { + return nil, ErrCodexQualityUnavailable + } + operation = canonical + id, valid = canonicalCodexQualityID(id) + if !valid { + return nil, ErrCodexQualityUnavailable + } + rt, err := s.codexQualityRuntime() + if err != nil { + return nil, err + } + ctx = rt.ctx(ctx) + ctx, cancel := context.WithTimeout(ctx, 65*time.Second) + defer cancel() + run, _, err := readCodexQualityRun(ctx, rt.store, id) + if err != nil || run.ActorID != actor || run.AccountID != accountID || !codexQualityActive(run, time.Now()) { + return nil, ErrCodexQualityUnavailable + } + key, err := lookup(ctx, run.APIKeyID) + if err != nil || !codexQualityKeyUsable(key, actor, run.APIKeyID, run.GroupID) { + return nil, ErrCodexQualityUnavailable + } + for _, a := range run.Attempts { + if a.OperationID == operation { + return s.ReadCodexQualityRun(ctx, actor, accountID, id) + } + } + account, err := rt.account(ctx, run) + if err != nil { + return nil, err + } + if s.concurrencyService == nil { + return nil, ErrCodexQualityUnavailable + } + concurrency, err := s.concurrencyService.AcquireAccountSlot(ctx, accountID, account.Concurrency) + if err != nil || concurrency == nil || !concurrency.Acquired { + return nil, ErrCodexQualityUnavailable + } + defer concurrency.ReleaseFunc() + lease := extensionv1.LeaseRequest{Namespace: "routing-account", Key: fmt.Sprint(accountID), Owner: uuid.NewString(), TTLSeconds: 75} + locked, err := rt.store.AcquireExtensionLease(ctx, codexRuntimePluginKey, lease) + if err != nil || !locked.Acquired { + return nil, ErrCodexQualityUnavailable + } + lease.Generation = locked.Generation + defer func() { + release, stop := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second) + defer stop() + _, _ = rt.store.ReleaseExtensionLease(release, codexRuntimePluginKey, lease) + }() + query := extensionv1.CodexRoutingQuery{AccountID: accountID, Model: codexQualityModel, Transport: "http", Stage: "acquire", OperationID: "quality-" + id + "-" + operation, Scope: &run.Scope} + call := func() (extensionv1.CodexRoutingProbeResult, error) { + e := &codexQualityExecution{runtime: rt, runID: id, operationID: operation, stage: query.Stage, accountID: accountID, keyLookup: lookup} + stageCtx := context.WithValue(ctx, codexQualityExecutionKey{}, e) + raw, _ := json.Marshal(query) + result, callErr := rt.host.Call(stageCtx, extensionv1.HostInvocation{Operation: extensionv1.HostCodexRoutingProbe, Payload: raw}) + var out extensionv1.CodexRoutingProbeResult + if callErr != nil || result.Code != "" || json.Unmarshal(result.Payload, &out) != nil { + return out, ErrCodexQualityUnavailable + } + return out, nil + } + acquired, err := call() + if err != nil || !acquired.Valid || acquired.Bundle == nil { + return s.ReadCodexQualityRun(ctx, actor, accountID, id) + } + query.Stage, query.Bundle = "verify", acquired.Bundle + verified, err := call() + if err != nil || !verified.Valid || verified.Bundle == nil || !verified.Observation.Completed || !verified.Observation.ModelMatched || verified.Observation.ResponseModel != codexQualityModel || verified.Scope.ConnectionLeaseID == "" || !run.Scope.SameOwner(verified.Scope) { + return s.ReadCodexQualityRun(ctx, actor, accountID, id) + } + q := &extensionv1.CodexRoutingQualification{Scope: verified.Scope, Bundle: *verified.Bundle, Model: codexQualityModel, VerifiedAt: time.Now().UTC(), ExpiresAt: verified.Bundle.ExpiresAt} + saved, err := mutateCodexQualityRun(ctx, rt.store, id, func(current *codexQualityRun) error { + if !codexQualityActive(*current, time.Now()) || current.ActorID != actor || !current.Scope.SameOwner(q.Scope) { + return ErrCodexQualityUnavailable + } + current.Qualification, current.RouteRuntimeGeneration = q, rt.installation.RuntimeGeneration + current.RouteGeneration++ + return nil + }) + if err != nil { + s.closeCodexQualityConnection(q) + return nil, err + } + // The old diagnostic connection has no ordinary qualification owners. + if run.Qualification != nil && run.Qualification.Scope.ConnectionLeaseID != q.Scope.ConnectionLeaseID { + s.closeCodexQualityConnection(run.Qualification) + } + return codexQualityView(saved, rt.installation.RuntimeGeneration), nil +} diff --git a/backend/internal/service/codex_quality_runtime.go b/backend/internal/service/codex_quality_runtime.go new file mode 100644 index 000000000000..49653447e627 --- /dev/null +++ b/backend/internal/service/codex_quality_runtime.go @@ -0,0 +1,480 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "slices" + "strings" + "time" + + extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" + "github.com/gin-gonic/gin" + "github.com/tidwall/gjson" +) + +type codexQualityRuntime struct { + s *OpenAIGatewayService + store PluginExtensionStateStore + installation *PluginInstallation + host *pluginExtensionHost +} + +type codexQualityExecutionKey struct{} +type codexQualityHeadersKey struct{} +type codexQualityHeaders struct { + grant, trial string + present bool +} +type codexQualityExecution struct { + runtime *codexQualityRuntime + runID, grantDigest, trialID, operationID, stage string + accountID int64 + requestModel, effort string + qualification *extensionv1.CodexRoutingQualification + keyLookup CodexQualityKeyLookup +} + +type CodexQualityKeyLookup func(context.Context, int64) (*APIKey, error) + +func codexQualityKeyUsable(key *APIKey, actor, id, group int64) bool { + return key != nil && key.ID == id && key.UserID == actor && key.GroupID != nil && *key.GroupID == group && key.IsActive() && !key.IsExpired() && !key.IsQuotaExhausted() +} + +func codexQualityExecutionFromContext(ctx context.Context) *codexQualityExecution { + if ctx == nil { + return nil + } + e, _ := ctx.Value(codexQualityExecutionKey{}).(*codexQualityExecution) + return e +} + +func IsCodexQualityRequest(ctx context.Context) bool { + return codexQualityExecutionFromContext(ctx) != nil +} + +// Remove the capability before any request copying, forwarding or diagnostics. +func StageCodexQualityHeaders(c *gin.Context) { + if c == nil || c.Request == nil { + return + } + h := codexQualityHeaders{} + grantCount, trialCount := 0, 0 + for key, values := range c.Request.Header { + if strings.EqualFold(key, CodexQualityGrantHeader) || strings.EqualFold(key, CodexQualityTrialHeader) { + h.present = true + value := "" + if len(values) == 1 { + value = values[0] + } + if strings.EqualFold(key, CodexQualityGrantHeader) { + h.grant = value + grantCount += len(values) + } else { + h.trial = value + trialCount += len(values) + } + delete(c.Request.Header, key) + } + } + if grantCount != 1 || trialCount != 1 { + h.grant, h.trial = "", "" + } + if h.present { + c.Request = c.Request.WithContext(context.WithValue(c.Request.Context(), codexQualityHeadersKey{}, h)) + } +} + +func (s *OpenAIGatewayService) codexQualityRuntime() (*codexQualityRuntime, error) { + if s == nil || s.pluginManager == nil || s.accountRepo == nil { + return nil, ErrCodexQualityUnavailable + } + i, _ := s.pluginManager.installedByKey(codexRuntimePluginKey) + store, ok := s.pluginManager.repo.(PluginExtensionStateStore) + if !ok || i == nil { + return nil, ErrCodexQualityUnavailable + } + host, ok := s.pluginManager.buildHostServices(i).(*pluginHostServiceServer) + if !ok || host == nil || host.extension == nil { + return nil, ErrCodexQualityUnavailable + } + broker, ok := host.extension.(*pluginExtensionHost) + if !ok || broker == nil { + return nil, ErrCodexQualityUnavailable + } + return &codexQualityRuntime{s: s, store: store, installation: i, host: broker}, nil +} + +func (rt *codexQualityRuntime) ctx(ctx context.Context) context.Context { + return WithPluginExecution(ctx, rt.installation) +} + +func (rt *codexQualityRuntime) scope(ctx context.Context, id int64) (extensionv1.CodexRoutingScope, error) { + raw, _ := json.Marshal(extensionv1.CodexRoutingQuery{AccountID: id, Transport: "http"}) + result, err := rt.host.Call(rt.ctx(ctx), extensionv1.HostInvocation{Operation: extensionv1.HostCodexRoutingScope, Payload: raw}) + var scope extensionv1.CodexRoutingScope + if err != nil || result.Code != "" || json.Unmarshal(result.Payload, &scope) != nil || scope.AccountID != id { + return scope, ErrCodexQualityUnavailable + } + return scope, nil +} + +func qualityProxyID(a *Account) int64 { + if a.ProxyID != nil { + return *a.ProxyID + } + return 0 +} + +func (rt *codexQualityRuntime) account(ctx context.Context, run codexQualityRun) (*Account, error) { + a, err := rt.s.accountRepo.GetByID(ctx, run.AccountID) + // A diagnostic grant never admits an account enabled for ordinary traffic. + // Recheck the authoritative row here, including immediately before each send. + if err != nil || a == nil || a.Schedulable || a.ID != run.AccountID || a.Platform != PlatformOpenAI || a.Type != AccountTypeOAuth || a.IsShadow() || !slices.Contains(a.GroupIDs, run.GroupID) || qualityProxyID(a) != run.ProxyID || CodexTicketAccountIdentity(a) != run.Scope.Identity { + return nil, ErrCodexQualityUnavailable + } + plan := strings.ToLower(strings.TrimSpace(a.GetCredential("plan_type"))) + if plan != "pro" && plan != "chatgpt_pro" { + return nil, ErrCodexQualityUnavailable + } + if rt.s.checkChannelPricingRestriction(ctx, &run.GroupID, codexQualityModel) || (rt.s.needsUpstreamChannelRestrictionCheck(ctx, &run.GroupID) && rt.s.isUpstreamModelRestrictedByChannel(ctx, run.GroupID, a, codexQualityModel, false)) { + return nil, ErrCodexQualityUnavailable + } + // This private stack copy changes only the manual switch for eligibility. + // It is never returned, cached, projected, or persisted. + check := *a + check.Schedulable = true + if eligible, _ := openAICompatibleAccountEligibilityBeforeProfit(ctx, &check, PlatformOpenAI, codexQualityModel, false, OpenAIEndpointCapabilityResponses); !eligible { + return nil, ErrCodexQualityUnavailable + } + if rt.s.codexQualityHealthBlocked(ctx, &check) { + return nil, ErrCodexQualityUnavailable + } + if rt.s.openAIGroupRequiresPrivacySet(ctx, &run.GroupID) && !a.IsPrivacySet() { + return nil, ErrCodexQualityUnavailable + } + scope, err := rt.scope(ctx, a.ID) + if err != nil || !run.Scope.SameOwner(scope) { + return nil, ErrCodexQualityUnavailable + } + return a, nil +} + +func (s *OpenAIGatewayService) CreateCodexQualityRun(ctx context.Context, actor, accountID int64, key *APIKey, request CodexQualityCreateRequest) (*CodexQualityRunView, error) { + id, validID := canonicalCodexQualityID(request.RunID) + if !validID || !validCodexQualityHash(request.PromptSHA256) || actor <= 0 || key == nil || key.GroupID == nil || !codexQualityKeyUsable(key, actor, request.APIKeyID, *key.GroupID) { + return nil, ErrCodexQualityUnavailable + } + request.RunID = id + if request.MaxSends == 0 { + request.MaxSends = codexQualityMaxSends + } + if request.TTLSeconds == 0 { + request.TTLSeconds = 7200 + } + if request.MaxSends < 1 || request.MaxSends > codexQualityMaxSends || request.TTLSeconds < 60 || request.TTLSeconds > 7200 { + return nil, ErrCodexQualityUnavailable + } + rt, err := s.codexQualityRuntime() + if err != nil { + return nil, err + } + ctx = rt.ctx(ctx) + a, err := s.accountRepo.GetByID(ctx, accountID) + if err != nil || a == nil { + return nil, ErrCodexQualityUnavailable + } + scope, err := rt.scope(ctx, accountID) + if err != nil { + return nil, err + } + wanted := codexQualityRun{RunID: request.RunID, ActorID: actor, APIKeyID: key.ID, AccountID: accountID, GroupID: *key.GroupID, ProxyID: qualityProxyID(a), Scope: scope, PromptSHA256: request.PromptSHA256, MaxSends: request.MaxSends, Status: "open", Attempts: []CodexQualityAttempt{}} + if _, err = rt.account(ctx, wanted); err != nil { + return nil, err + } + grant, digest, err := newCodexQualityGrant(request.RunID) + if err != nil { + return nil, err + } + // Reissuing a grant invalidates the old capability; spent entries survive. + run, err := issueCodexQualityGrant(ctx, rt.store, wanted, digest, time.Now().UTC().Add(time.Duration(request.TTLSeconds)*time.Second)) + if err != nil { + return nil, err + } + view := codexQualityView(run, rt.installation.RuntimeGeneration) + view.Grant = grant + return view, nil +} + +func (s *OpenAIGatewayService) ReadCodexQualityRun(ctx context.Context, actor, accountID int64, id string) (*CodexQualityRunView, error) { + rt, err := s.codexQualityRuntime() + if err != nil { + return nil, err + } + run, _, err := readCodexQualityRun(rt.ctx(ctx), rt.store, id) + if err != nil || run.RunID == "" || run.ActorID != actor || run.AccountID != accountID { + return nil, ErrCodexQualityUnavailable + } + view := codexQualityView(run, rt.installation.RuntimeGeneration) + if view.RouteReady { + if _, err = rt.account(ctx, run); err != nil { + view.RouteReady = false + } else if _, err = rt.qualification(ctx, run); err != nil { + view.RouteReady = false + } + } + return view, nil +} + +func (s *OpenAIGatewayService) CloseCodexQualityRun(ctx context.Context, actor, accountID int64, id string) (*CodexQualityRunView, error) { + rt, err := s.codexQualityRuntime() + if err != nil { + return nil, err + } + run, err := mutateCodexQualityRun(rt.ctx(ctx), rt.store, id, func(run *codexQualityRun) error { + if run.RunID == "" || run.ActorID != actor || run.AccountID != accountID { + return ErrCodexQualityUnavailable + } + run.Status, run.GrantDigest = "closed", "" + return nil + }) + if err != nil { + return nil, err + } + if run.Qualification != nil { + s.closeCodexQualityConnection(run.Qualification) + } + rt.clearClosedMaterial(ctx, run) + return codexQualityView(run, rt.installation.RuntimeGeneration), nil +} + +func (s *OpenAIGatewayService) AdmitCodexQualityRequest(c *gin.Context, key *APIKey, body []byte, lookups ...CodexQualityKeyLookup) error { + h, exists := c.Request.Context().Value(codexQualityHeadersKey{}).(codexQualityHeaders) + if !exists { + return nil + } + if len(lookups) != 1 || lookups[0] == nil { + return ErrCodexQualityUnavailable + } + id, secret, ok := strings.Cut(h.grant, ".") + id, validID := canonicalCodexQualityID(id) + trial, validTrial := canonicalCodexQualityID(h.trial) + if !ok || len(secret) != 43 || !validID || !validTrial || c.Request.Method != http.MethodPost || c.Request.URL.Path != "/v1/responses" || key == nil || key.GroupID == nil { + return ErrCodexQualityUnavailable + } + h.trial = trial + var fields map[string]json.RawMessage + if !validCodexQualityJSON(body) || json.Unmarshal(body, &fields) != nil { + return ErrCodexQualityUnavailable + } + for name := range fields { + if !slices.Contains([]string{"model", "input", "reasoning", "stream", "store", "prompt_cache_key", "client_metadata"}, name) { + return ErrCodexQualityUnavailable + } + } + input := gjson.GetBytes(body, "input") + if input.Type != gjson.String || gjson.GetBytes(body, "model").String() != codexQualityModel || gjson.GetBytes(body, "reasoning.effort").String() != codexQualityEffort || !gjson.GetBytes(body, "stream").Bool() || gjson.GetBytes(body, "store").Bool() { + return ErrCodexQualityUnavailable + } + rt, err := s.codexQualityRuntime() + if err != nil { + return err + } + run, _, err := readCodexQualityRun(rt.ctx(c.Request.Context()), rt.store, id) + digest := codexQualityHash(h.grant) + if err != nil || !codexQualityActive(run, time.Now()) || run.APIKeyID != key.ID || run.ActorID != key.UserID || run.GroupID != *key.GroupID || run.PromptSHA256 != codexQualityHash(input.String()) || !codexQualityGrantMatches(run, digest) { + return ErrCodexQualityUnavailable + } + for _, old := range run.Attempts { + if old.Stage == "business" && old.TrialID == h.trial { + return ErrCodexQualitySpent + } + } + ctx := ensureOpenAIRuntimeBreakerProbeOwner(c.Request.Context()) + if _, err = rt.account(ctx, run); err != nil { + return err + } + q, err := rt.qualification(ctx, run) + if err != nil { + return err + } + e := &codexQualityExecution{runtime: rt, runID: id, grantDigest: digest, trialID: h.trial, stage: "business", accountID: run.AccountID, qualification: q, keyLookup: lookups[0]} + c.Request = c.Request.WithContext(context.WithValue(ctx, codexQualityExecutionKey{}, e)) + return nil +} + +// Reject duplicate object members: gjson and an upstream JSON parser must not +// disagree about the prompt or effort authorized by the capability. +func validCodexQualityJSON(body []byte) bool { + if len(body) > 1<<20 { + return false + } + decoder := json.NewDecoder(bytes.NewReader(body)) + var value func(int) bool + value = func(depth int) bool { + if depth > 16 { + return false + } + token, err := decoder.Token() + if err != nil { + return false + } + delim, ok := token.(json.Delim) + if !ok { + return true + } + switch delim { + case '{': + seen := map[string]bool{} + for decoder.More() { + key, err := decoder.Token() + name, ok := key.(string) + if err != nil || !ok || seen[name] { + return false + } + seen[name] = true + if !value(depth + 1) { + return false + } + } + case '[': + for decoder.More() { + if !value(depth + 1) { + return false + } + } + default: + return false + } + end, err := decoder.Token() + return err == nil && ((delim == '{' && end == json.Delim('}')) || (delim == '[' && end == json.Delim(']'))) + } + if !value(0) { + return false + } + _, err := decoder.Token() + return err == io.EOF +} + +func (rt *codexQualityRuntime) qualification(ctx context.Context, run codexQualityRun) (*extensionv1.CodexRoutingQualification, error) { + q := run.Qualification + if !codexQualityActive(run, time.Now()) || q == nil || run.RouteRuntimeGeneration != rt.installation.RuntimeGeneration || !q.Valid(time.Now(), run.AccountID, run.Scope.Identity, codexQualityModel) || !q.Scope.SameOwner(run.Scope) || q.Scope.Transport != "http" || q.Scope.ConnectionLeaseID == "" { + return nil, ErrCodexQualityUnavailable + } + bundle, err := readCodexRoutingBundle(rt.ctx(ctx), rt.store, codexRuntimePluginKey, q.Bundle, q.Scope, true) + if err != nil || bundle.Model != codexQualityModel || bundle.Scope.ConnectionLeaseID != q.Scope.ConnectionLeaseID { + return nil, ErrCodexQualityUnavailable + } + if leases, ok := rt.s.httpUpstream.(interface { + HasCodexQualityConnection(string, int64, string) bool + }); !ok || !leases.HasCodexQualityConnection(q.Scope.ConnectionLeaseID, run.AccountID, codexRoutingScopeKey(q.Scope)) { + return nil, ErrCodexQualityUnavailable + } + return q, nil +} + +func (e *codexQualityExecution) current(ctx context.Context) (codexQualityRun, error) { + run, _, err := readCodexQualityRun(e.runtime.ctx(ctx), e.runtime.store, e.runID) + if err != nil || !codexQualityActive(run, time.Now()) || run.AccountID != e.accountID || (e.grantDigest != "" && !codexQualityGrantMatches(run, e.grantDigest)) { + return run, ErrCodexQualityUnavailable + } + if e.keyLookup == nil { + return run, ErrCodexQualityUnavailable + } + key, err := e.keyLookup(ctx, run.APIKeyID) + if err != nil || !codexQualityKeyUsable(key, run.ActorID, run.APIKeyID, run.GroupID) { + return run, ErrCodexQualityUnavailable + } + return run, nil +} + +func (s *OpenAIGatewayService) SelectCodexQualityAccount(ctx context.Context) (*AccountSelectionResult, error) { + e := codexQualityExecutionFromContext(ctx) + if e == nil { + return nil, ErrCodexQualityUnavailable + } + run, err := e.current(ctx) + if err != nil { + return nil, err + } + a, err := e.runtime.account(ctx, run) + if err != nil { + return nil, err + } + if _, err = e.runtime.qualification(ctx, run); err != nil { + return nil, err + } + if s.concurrencyService == nil { + return nil, ErrCodexQualityUnavailable + } + acquired, err := s.concurrencyService.AcquireAccountSlot(ctx, a.ID, a.Concurrency) + if err != nil || acquired == nil || !acquired.Acquired { + return nil, ErrCodexQualityUnavailable + } + return attachSelectionProfitGate(ctx, attachSelectionRuntimeBreakerProbe(ctx, &AccountSelectionResult{Account: a, Acquired: true, ReleaseFunc: acquired.ReleaseFunc})), nil +} + +func codexQualityRequestQualification(ctx context.Context, account *Account, model string) (*extensionv1.CodexRoutingQualification, *PluginInstallation, error) { + e := codexQualityExecutionFromContext(ctx) + if e == nil || account == nil || account.ID != e.accountID || model != codexQualityModel { + return nil, nil, ErrCodexQualityUnavailable + } + run, err := e.current(ctx) + if err != nil { + return nil, nil, err + } + if _, err = e.runtime.account(ctx, run); err != nil { + return nil, nil, err + } + q, err := e.runtime.qualification(ctx, run) + return q, e.runtime.installation, err +} + +func codexQualityBundleKey(ctx context.Context, kind string) string { + if e := codexQualityExecutionFromContext(ctx); e != nil { + return "bundle.quality." + codexQualityHash(e.runID)[:32] + "." + kind + } + return "" +} + +func codexQualityClockKey(ctx context.Context, fallback string) string { + if e := codexQualityExecutionFromContext(ctx); e != nil { + return "clock.quality." + codexQualityHash(e.runID)[:32] + } + return fallback +} + +func (s *OpenAIGatewayService) closeCodexQualityConnection(q *extensionv1.CodexRoutingQualification) { + if closer, ok := s.httpUpstream.(interface{ CloseCodexQualityConnection(string, int64, string) }); ok && q != nil { + closer.CloseCodexQualityConnection(q.Scope.ConnectionLeaseID, q.Scope.AccountID, codexRoutingScopeKey(q.Scope)) + } +} + +func (rt *codexQualityRuntime) clearClosedMaterial(ctx context.Context, run codexQualityRun) { + ctx, cancel := context.WithTimeout(rt.ctx(context.WithoutCancel(ctx)), 3*time.Second) + defer cancel() + for _, kind := range []string{"candidate", "live"} { + key := "bundle.quality." + codexQualityHash(run.RunID)[:32] + "." + kind + record, err := rt.store.ReadExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: key}) + var bundle codexRoutingPrivateBundle + if err != nil || !record.Found || json.Unmarshal(record.Value, &bundle) != nil || !bundle.Scope.SameOwner(run.Scope) { + continue + } + for i := range bundle.Cookies { + bundle.Cookies[i].Value = "" + } + bundle.Status = "expired" + raw, _ := json.Marshal(bundle) + _, _ = rt.store.CompareSwapExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: key, ExpectedRevision: record.Revision, Value: raw}) + } + key := "clock.quality." + codexQualityHash(run.RunID)[:32] + record, err := rt.store.ReadExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: key}) + var clock codexRoutingCookieClock + if err == nil && record.Found && json.Unmarshal(record.Value, &clock) == nil && clock.Scope.SameOwner(run.Scope) { + clock.Cookies = nil + raw, _ := json.Marshal(clock) + _, _ = rt.store.CompareSwapExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: key, ExpectedRevision: record.Revision, Value: raw}) + } +} diff --git a/backend/internal/service/codex_quality_state.go b/backend/internal/service/codex_quality_state.go new file mode 100644 index 000000000000..7c5f3212e216 --- /dev/null +++ b/backend/internal/service/codex_quality_state.go @@ -0,0 +1,266 @@ +package service + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "strings" + "time" + + extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" + "github.com/google/uuid" +) + +const ( + CodexQualityGrantHeader = "X-Sub2API-Quality-Grant" + CodexQualityTrialHeader = "X-Sub2API-Quality-Trial" + codexQualityModel = "gpt-6-astra" + codexQualityEffort = "high" + codexQualityMaxSends = 60 +) + +// A diagnostic failure is deliberately small and never contains upstream text. +var ErrCodexQualityUnavailable = errors.New("codex_quality_unavailable") +var ErrCodexQualitySpent = errors.New("codex_quality_attempt_spent") + +type CodexQualityCreateRequest struct { + RunID string `json:"run_id"` + APIKeyID int64 `json:"api_key_id"` + PromptSHA256 string `json:"prompt_sha256"` + MaxSends int `json:"max_sends"` + TTLSeconds int `json:"ttl_seconds"` +} + +type CodexQualityAttempt struct { + Stage string `json:"stage"` + TrialID string `json:"trial_id,omitempty"` + OperationID string `json:"operation_id,omitempty"` + AccountID int64 `json:"account_id"` + State string `json:"state"` + ReservedAt time.Time `json:"reserved_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + RequestModel string `json:"request_model"` + ReasoningEffort string `json:"reasoning_effort"` + ResponseModels []string `json:"response_models"` + CreatedModel *string `json:"created_model"` + TerminalModel *string `json:"terminal_model"` + HeaderModel *string `json:"header_model"` + HeaderModels []string `json:"header_models"` + Completed bool `json:"completed"` + HTTPStatus int `json:"http_status,omitempty"` + ErrorCode string `json:"error_code,omitempty"` + ReasoningTokens *int64 `json:"reasoning_tokens"` + InputTokens *int64 `json:"input_tokens"` + OutputTokens *int64 `json:"output_tokens"` + ConnectionFingerprint string `json:"connection_fingerprint,omitempty"` + QualificationFingerprint string `json:"qualification_fingerprint,omitempty"` +} + +type CodexQualityRunView struct { + RunID string `json:"run_id"` + Grant string `json:"grant,omitempty"` + AccountID int64 `json:"account_id"` + APIKeyID int64 `json:"api_key_id"` + GroupID int64 `json:"group_id"` + Model string `json:"model"` + ReasoningEffort string `json:"reasoning_effort"` + PromptSHA256 string `json:"prompt_sha256"` + Status string `json:"status"` + MaxSends int `json:"max_sends"` + UsedSends int `json:"used_sends"` + ExpiresAt time.Time `json:"expires_at"` + RouteReady bool `json:"route_ready"` + RouteExpiresAt *time.Time `json:"route_expires_at"` + RouteGeneration int64 `json:"route_generation"` + ConnectionFingerprint string `json:"connection_fingerprint,omitempty"` + Attempts []CodexQualityAttempt `json:"attempts"` +} + +// Stored only in the existing host-private namespace. No grant, key, prompt, +// response text, Cookie or STATE value is stored in this ledger. +type codexQualityRun struct { + RunID string `json:"run_id"` + ActorID int64 `json:"actor_id"` + APIKeyID int64 `json:"api_key_id"` + AccountID int64 `json:"account_id"` + GroupID int64 `json:"group_id"` + ProxyID int64 `json:"proxy_id"` + Scope extensionv1.CodexRoutingScope `json:"scope"` + PromptSHA256 string `json:"prompt_sha256"` + GrantDigest string `json:"grant_digest"` + Status string `json:"status"` + MaxSends int `json:"max_sends"` + UsedSends int `json:"used_sends"` + ExpiresAt time.Time `json:"expires_at"` + RouteGeneration int64 `json:"route_generation"` + RouteRuntimeGeneration int64 `json:"route_runtime_generation"` + Qualification *extensionv1.CodexRoutingQualification `json:"qualification,omitempty"` + Attempts []CodexQualityAttempt `json:"attempts"` +} + +func codexQualityKey(id string) string { return "quality-run." + id } + +func canonicalCodexQualityID(text string) (string, bool) { + if len(text) != 36 { + return "", false + } + id, err := uuid.Parse(text) + if err != nil { + return "", false + } + return id.String(), true +} +func codexQualityHash(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + +func readCodexQualityRun(ctx context.Context, store PluginExtensionStateStore, id string) (codexQualityRun, int64, error) { + var run codexQualityRun + canonical, valid := canonicalCodexQualityID(id) + if !valid || store == nil { + return run, 0, ErrCodexQualityUnavailable + } + id = canonical + record, err := store.ReadExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: codexQualityKey(id)}) + if err != nil { + return run, 0, ErrCodexQualityUnavailable + } + if !record.Found { + return run, 0, nil + } + if json.Unmarshal(record.Value, &run) != nil || run.RunID != id || run.MaxSends < 1 || run.MaxSends > codexQualityMaxSends || run.UsedSends != len(run.Attempts) || run.UsedSends > run.MaxSends { + return codexQualityRun{}, 0, ErrCodexQualityUnavailable + } + return run, record.Revision, nil +} + +func mutateCodexQualityRun(ctx context.Context, store PluginExtensionStateStore, id string, change func(*codexQualityRun) error) (codexQualityRun, error) { + canonical, valid := canonicalCodexQualityID(id) + if !valid { + return codexQualityRun{}, ErrCodexQualityUnavailable + } + id = canonical + for attempt := 0; attempt < 12; attempt++ { + run, revision, err := readCodexQualityRun(ctx, store, id) + if err != nil { + return run, err + } + if err = change(&run); err != nil { + return run, err + } + raw, err := json.Marshal(run) + if err != nil { + return run, ErrCodexQualityUnavailable + } + result, err := store.CompareSwapExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: codexQualityKey(id), ExpectedRevision: revision, Value: raw}) + if err != nil { + return run, ErrCodexQualityUnavailable + } + if result.Applied { + return run, nil + } + } + return codexQualityRun{}, ErrCodexQualityUnavailable +} + +func newCodexQualityGrant(id string) (string, string, error) { + var secret [32]byte + if _, err := rand.Read(secret[:]); err != nil { + return "", "", ErrCodexQualityUnavailable + } + grant := id + "." + base64.RawURLEncoding.EncodeToString(secret[:]) + return grant, codexQualityHash(grant), nil +} + +func codexQualityGrantMatches(run codexQualityRun, digest string) bool { + return len(digest) == 64 && len(run.GrantDigest) == 64 && subtle.ConstantTimeCompare([]byte(digest), []byte(run.GrantDigest)) == 1 +} + +func codexQualityActive(run codexQualityRun, now time.Time) bool { + return run.RunID != "" && run.Status == "open" && now.Before(run.ExpiresAt) +} + +func issueCodexQualityGrant(ctx context.Context, store PluginExtensionStateStore, wanted codexQualityRun, digest string, expiry time.Time) (codexQualityRun, error) { + canonical, valid := canonicalCodexQualityID(wanted.RunID) + if !valid { + return codexQualityRun{}, ErrCodexQualityUnavailable + } + wanted.RunID = canonical + return mutateCodexQualityRun(ctx, store, wanted.RunID, func(run *codexQualityRun) error { + if run.RunID == "" { + *run = wanted + } else if run.Status == "closed" || run.ActorID != wanted.ActorID || run.APIKeyID != wanted.APIKeyID || run.AccountID != wanted.AccountID || run.GroupID != wanted.GroupID || run.ProxyID != wanted.ProxyID || run.PromptSHA256 != wanted.PromptSHA256 || run.MaxSends != wanted.MaxSends || !run.Scope.SameOwner(wanted.Scope) { + return ErrCodexQualityUnavailable + } + run.GrantDigest, run.ExpiresAt = digest, expiry + return nil + }) +} + +func sameCodexQualityAttempt(a, b CodexQualityAttempt) bool { + return a.Stage == b.Stage && a.TrialID == b.TrialID && a.OperationID == b.OperationID +} + +func reserveCodexQualityAttempt(ctx context.Context, store PluginExtensionStateStore, id, grantDigest string, attempt CodexQualityAttempt) error { + _, err := mutateCodexQualityRun(ctx, store, id, func(run *codexQualityRun) error { + if !codexQualityActive(*run, time.Now()) || (grantDigest != "" && !codexQualityGrantMatches(*run, grantDigest)) || run.AccountID != attempt.AccountID { + return ErrCodexQualityUnavailable + } + for _, previous := range run.Attempts { + if sameCodexQualityAttempt(previous, attempt) { + return ErrCodexQualitySpent + } + } + if run.UsedSends >= run.MaxSends { + return ErrCodexQualitySpent + } + attempt.State, attempt.ReservedAt = "unknown", time.Now().UTC() + attempt.ResponseModels = []string{} + attempt.HeaderModels = []string{} + run.Attempts = append(run.Attempts, attempt) + run.UsedSends++ + return nil + }) + return err +} + +func finishCodexQualityAttempt(ctx context.Context, store PluginExtensionStateStore, id string, finished CodexQualityAttempt) { + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second) + defer cancel() + _, _ = mutateCodexQualityRun(ctx, store, id, func(run *codexQualityRun) error { + for index, old := range run.Attempts { + if sameCodexQualityAttempt(old, finished) { + finished.ReservedAt = old.ReservedAt + now := time.Now().UTC() + finished.FinishedAt = &now + run.Attempts[index] = finished + return nil + } + } + return ErrCodexQualityUnavailable + }) +} + +func codexQualityView(run codexQualityRun, generation int64) *CodexQualityRunView { + v := &CodexQualityRunView{RunID: run.RunID, AccountID: run.AccountID, APIKeyID: run.APIKeyID, GroupID: run.GroupID, Model: codexQualityModel, ReasoningEffort: codexQualityEffort, PromptSHA256: run.PromptSHA256, Status: run.Status, MaxSends: run.MaxSends, UsedSends: run.UsedSends, ExpiresAt: run.ExpiresAt, RouteGeneration: run.RouteGeneration, Attempts: run.Attempts} + if v.Attempts == nil { + v.Attempts = []CodexQualityAttempt{} + } + if q := run.Qualification; q != nil { + v.RouteExpiresAt = &q.ExpiresAt + v.RouteReady = codexQualityActive(run, time.Now()) && generation == run.RouteRuntimeGeneration && q.Valid(time.Now(), run.AccountID, run.Scope.Identity, codexQualityModel) + v.ConnectionFingerprint = codexQualityHash(q.Scope.ConnectionLeaseID)[:16] + } + return v +} + +func validCodexQualityHash(value string) bool { + decoded, err := hex.DecodeString(value) + return err == nil && len(decoded) == 32 && strings.ToLower(value) == value +} diff --git a/backend/internal/service/codex_quality_test.go b/backend/internal/service/codex_quality_test.go new file mode 100644 index 000000000000..b6bc76615c80 --- /dev/null +++ b/backend/internal/service/codex_quality_test.go @@ -0,0 +1,287 @@ +package service + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +func qualityStateFixture(t *testing.T, limit int) (*routingMemoryStore, codexQualityRun) { + t.Helper() + store := &routingMemoryStore{} + run := codexQualityRun{RunID: uuid.NewString(), ActorID: 9, APIKeyID: 101, AccountID: 16380, GroupID: 53, ProxyID: 34, Scope: extensionv1.CodexRoutingScope{AccountID: 16380, Identity: "owner", ProfileHash: "profile", RouteHash: "route", Transport: "http"}, PromptSHA256: codexQualityHash("question"), MaxSends: limit, Status: "open"} + _, digest, err := newCodexQualityGrant(run.RunID) + require.NoError(t, err) + run, err = issueCodexQualityGrant(context.Background(), store, run, digest, time.Now().Add(time.Hour)) + require.NoError(t, err) + return store, run +} + +func TestCodexQualityBudgetReplayAndConcurrentCAS(t *testing.T) { + store, run := qualityStateFixture(t, 8) + ctx := context.Background() + first := CodexQualityAttempt{Stage: "business", TrialID: uuid.NewString(), AccountID: run.AccountID} + require.NoError(t, reserveCodexQualityAttempt(ctx, store, run.RunID, run.GrantDigest, first)) + require.ErrorIs(t, reserveCodexQualityAttempt(ctx, store, run.RunID, run.GrantDigest, first), ErrCodexQualitySpent) + var successes atomic.Int32 + var workers sync.WaitGroup + for i := 0; i < 24; i++ { + workers.Add(1) + go func() { + defer workers.Done() + attempt := CodexQualityAttempt{Stage: "business", TrialID: uuid.NewString(), AccountID: run.AccountID} + if reserveCodexQualityAttempt(ctx, store, run.RunID, run.GrantDigest, attempt) == nil { + successes.Add(1) + } + }() + } + workers.Wait() + saved, _, err := readCodexQualityRun(ctx, store, run.RunID) + require.NoError(t, err) + require.Equal(t, int(successes.Load())+1, saved.UsedSends) + require.LessOrEqual(t, saved.UsedSends, 8) + require.Equal(t, saved.UsedSends, len(saved.Attempts)) + for _, a := range saved.Attempts { + require.Equal(t, "unknown", a.State) + } +} + +func TestCodexQualityGrantReissueKeepsBudgetAndScope(t *testing.T) { + store, run := qualityStateFixture(t, 3) + ctx := context.Background() + for _, stage := range []string{"acquire", "verify", "business"} { + require.NoError(t, reserveCodexQualityAttempt(ctx, store, run.RunID, run.GrantDigest, CodexQualityAttempt{Stage: stage, TrialID: "one", OperationID: "route", AccountID: run.AccountID})) + } + _, digest, err := newCodexQualityGrant(run.RunID) + require.NoError(t, err) + saved, err := issueCodexQualityGrant(ctx, store, run, digest, time.Now().Add(2*time.Hour)) + require.NoError(t, err) + require.Equal(t, 3, saved.UsedSends) + require.False(t, codexQualityGrantMatches(saved, run.GrantDigest)) + require.True(t, codexQualityGrantMatches(saved, digest)) + require.ErrorIs(t, reserveCodexQualityAttempt(ctx, store, run.RunID, digest, CodexQualityAttempt{Stage: "business", TrialID: "two", AccountID: run.AccountID}), ErrCodexQualitySpent) + for _, mutate := range []func(*codexQualityRun){func(r *codexQualityRun) { r.ActorID++ }, func(r *codexQualityRun) { r.APIKeyID++ }, func(r *codexQualityRun) { r.AccountID++ }, func(r *codexQualityRun) { r.GroupID++ }, func(r *codexQualityRun) { r.ProxyID++ }, func(r *codexQualityRun) { r.PromptSHA256 = codexQualityHash("changed") }, func(r *codexQualityRun) { r.MaxSends++ }, func(r *codexQualityRun) { r.Scope.Identity = "different" }} { + other := run + mutate(&other) + _, err := issueCodexQualityGrant(ctx, store, other, digest, time.Now().Add(time.Hour)) + require.ErrorIs(t, err, ErrCodexQualityUnavailable) + } + _, err = mutateCodexQualityRun(ctx, store, run.RunID, func(r *codexQualityRun) error { r.Status = "closed"; return nil }) + require.NoError(t, err) + _, err = issueCodexQualityGrant(ctx, store, run, digest, time.Now().Add(time.Hour)) + require.ErrorIs(t, err, ErrCodexQualityUnavailable) +} + +func TestCodexQualityRawObserverDoesNotRewriteOrInventUsage(t *testing.T) { + t.Run("event_header_disagrees_with_body", func(t *testing.T) { + raw := "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"model\":\"gpt-6-astra\",\"metadata\":{\"headers\":{\"openai-model\":\"gpt-6-luna\",\"x-openai-model\":[\"gpt-6-luna\"]}}}}\n\n" + var observed CodexQualityAttempt + body := &codexQualityObservedBody{ReadCloser: io.NopCloser(strings.NewReader(raw)), sse: true, attempt: CodexQualityAttempt{HTTPStatus: 200}, finish: func(a CodexQualityAttempt) { observed = a }} + recordCodexQualityHeaderModel(&body.attempt, "gpt-6-astra") + wire, err := io.ReadAll(body) + require.NoError(t, err) + require.Equal(t, raw, string(wire)) + require.Equal(t, []string{"gpt-6-astra"}, observed.ResponseModels) + require.Equal(t, []string{"gpt-6-astra", "gpt-6-luna"}, observed.HeaderModels) + require.Equal(t, "conflicting_models", *observed.HeaderModel) + }) + raw := "data: {\"type\":\"response.created\",\"response\":{\"model\":\"gpt-6-luna\"}}\n\n" + + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"29\"}\n\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"model\":\"gpt-6-luna\",\"usage\":{\"output_tokens\":0}}}\n\n" + var final CodexQualityAttempt + finishes := 0 + body := &codexQualityObservedBody{ReadCloser: io.NopCloser(strings.NewReader(raw)), sse: true, attempt: CodexQualityAttempt{HTTPStatus: 200}, finish: func(a CodexQualityAttempt) { final = a; finishes++ }} + forwarded, err := io.ReadAll(body) + require.NoError(t, err) + require.NoError(t, body.Close()) + require.Equal(t, raw, string(forwarded)) + require.Equal(t, 1, finishes) + require.True(t, final.Completed) + require.Equal(t, []string{"gpt-6-luna"}, final.ResponseModels) + require.Equal(t, "gpt-6-luna", *final.CreatedModel) + require.Equal(t, "gpt-6-luna", *final.TerminalModel) + require.Nil(t, final.ReasoningTokens) + require.Nil(t, final.InputTokens) + require.NotNil(t, final.OutputTokens) + require.Zero(t, *final.OutputTokens) +} + +func TestCodexQualityPrivateKeysAndHeaderScope(t *testing.T) { + require.True(t, validCodexQualityJSON([]byte(`{"model":"gpt-6-astra","input":"question","reasoning":{"effort":"high"}}`))) + for _, body := range []string{`{"input":"question","input":"changed"}`, `{"reasoning":{"effort":"high","effort":"low"}}`, `{} {}`} { + require.False(t, validCodexQualityJSON([]byte(body))) + } + ctx := context.WithValue(context.Background(), codexQualityExecutionKey{}, &codexQualityExecution{runID: uuid.NewString()}) + query := extensionv1.CodexRoutingQuery{AccountID: 16380, Model: codexQualityModel} + privateKey := fixedCodexRoutingBundleKey(ctx, query, "candidate") + require.LessOrEqual(t, len(privateKey), 80) + require.NotEqual(t, privateKey, fixedCodexRoutingBundleKey(context.Background(), query, "candidate")) + require.NotEqual(t, codexQualityClockKey(ctx, "clock.ordinary"), "clock.ordinary") + require.Equal(t, "clock.ordinary", codexQualityClockKey(context.Background(), "clock.ordinary")) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + StageCodexQualityHeaders(c) + require.NoError(t, (&OpenAIGatewayService{}).AdmitCodexQualityRequest(c, nil, nil), "ordinary requests are unchanged") + c.Request.Header.Set(CodexQualityGrantHeader, "invalid-secret") + c.Request.Header.Set(CodexQualityTrialHeader, uuid.NewString()) + StageCodexQualityHeaders(c) + require.Empty(t, c.Request.Header.Get(CodexQualityGrantHeader)) + require.Empty(t, c.Request.Header.Get(CodexQualityTrialHeader)) + require.ErrorIs(t, (&OpenAIGatewayService{}).AdmitCodexQualityRequest(c, nil, nil), ErrCodexQualityUnavailable) + require.False(t, IsCodexQualityRequest(c.Request.Context())) +} + +func TestCodexQualityLocalBudgetErrorIsNotAccountFailure(t *testing.T) { + ctx := context.WithValue(context.Background(), codexQualityExecutionKey{}, &codexQualityExecution{}) + service := &OpenAIGatewayService{} + for _, err := range []error{ErrCodexQualitySpent, ErrCodexQualityUnavailable} { + require.ErrorIs(t, service.handleOpenAIUpstreamTransportError(ctx, nil, nil, err, false), err) + } +} + +func TestCodexQualityUUIDCaseCannotCreateAnotherBudget(t *testing.T) { + store, run := qualityStateFixture(t, 3) + ctx := context.Background() + trial := "a1b2c3d4-1111-4111-8111-a1b2c3d4e5f6" + canonical, ok := canonicalCodexQualityID(strings.ToUpper(trial)) + require.True(t, ok) + require.Equal(t, trial, canonical) + first := CodexQualityAttempt{Stage: "business", TrialID: trial, AccountID: run.AccountID} + require.NoError(t, reserveCodexQualityAttempt(ctx, store, run.RunID, run.GrantDigest, first)) + first.TrialID = canonical + require.ErrorIs(t, reserveCodexQualityAttempt(ctx, store, strings.ToUpper(run.RunID), run.GrantDigest, first), ErrCodexQualitySpent) + other := run + other.RunID = strings.ToUpper(run.RunID) + saved, err := issueCodexQualityGrant(ctx, store, other, codexQualityHash("new grant"), time.Now().Add(time.Hour)) + require.NoError(t, err) + require.Equal(t, run.RunID, saved.RunID) + require.Equal(t, 1, saved.UsedSends) +} + +type qualityHealthRepository struct { + AccountRepository + errorWrites, cooldownWrites int +} + +func (r *qualityHealthRepository) SetError(context.Context, int64, string) error { + r.errorWrites++ + return nil +} +func (r *qualityHealthRepository) SetTempUnschedulable(context.Context, int64, time.Time, string) error { + r.cooldownWrites++ + return nil +} + +type quality403Counter struct{ resets int } + +func (*quality403Counter) IncrementOpenAI403Count(context.Context, int64, int) (int64, error) { + return 1, nil +} +func (c *quality403Counter) ResetOpenAI403Count(context.Context, int64) error { c.resets++; return nil } + +func TestCodexQualityHTTPFailuresAndSuccessDoNotChangeOrdinaryHealth(t *testing.T) { + repo := &qualityHealthRepository{} + counter := &quality403Counter{} + rateLimit := &RateLimitService{accountRepo: repo, openAI403CounterCache: counter} + s := &OpenAIGatewayService{accountRepo: repo, rateLimitService: rateLimit} + proxy := int64(34) + account := &Account{ID: 16380, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, ProxyID: &proxy} + ctx := context.WithValue(context.Background(), codexQualityExecutionKey{}, &codexQualityExecution{}) + for _, status := range []int{401, 429, 529} { + require.False(t, s.handleOpenAIAccountUpstreamError(ctx, account, status, nil, []byte(`{"error":{"code":"account_deactivated","message":"account deactivated"}}`), codexQualityModel)) + } + transportErr := errors.New("connection refused") + require.ErrorIs(t, s.handleOpenAIUpstreamTransportError(ctx, nil, account, transportErr, false), transportErr) + require.False(t, rateLimit.HandleStreamTimeout(ctx, account, codexQualityModel)) + s.observeOpenAIUsageAccountHealth(context.Background(), &OpenAIRecordUsageInput{Account: account, CodexQuality: true}) + require.Zero(t, repo.errorWrites) + require.Zero(t, repo.cooldownWrites) + require.Zero(t, counter.resets) + _, blocked := s.openaiAccountRuntimeBlockUntil.Load(account.ID) + require.False(t, blocked) + // The ordinary control retains both its permanent 401 observation and its + // successful-request reset; the diagnostic guard is not a global disable. + require.True(t, s.handleOpenAIAccountUpstreamError(context.Background(), account, 401, nil, []byte(`{"error":{"code":"account_deactivated","message":"account deactivated"}}`), codexQualityModel)) + require.Equal(t, 1, repo.errorWrites) + s.observeOpenAIUsageAccountHealth(context.Background(), &OpenAIRecordUsageInput{Account: account}) + require.Equal(t, 1, counter.resets) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + apiAccount := &Account{ID: 16463, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, ProxyID: &proxy} + _ = s.handleOpenAIUpstreamTransportError(WithOpenAIOfficialHTTPFailover(context.Background()), c, apiAccount, transportErr, false) + require.Equal(t, 1, repo.cooldownWrites) + // Proxy disconnect/success isolation is also independent of account health. + circuit := s.getOpenAIProxyStreamCircuit() + circuit.entries[proxy] = openAIProxyStreamCircuitEntry{failureCount: 1, blockedUntil: time.Now().Add(time.Minute)} + before := circuit.entries[proxy] + s.recordOpenAIProxyStreamDisconnect(account, transportErr, "fixture", ctx) + s.clearOpenAIProxyStreamDisconnect(account, ctx) + require.Equal(t, before, circuit.entries[proxy]) +} + +type qualityBreakerReadCache struct { + GatewayCache + reads int + blocked bool +} + +func (c *qualityBreakerReadCache) CodexQualityRuntimeBlocked(context.Context, int64, []string) (bool, error) { + c.reads++ + return c.blocked, nil +} +func (*qualityBreakerReadCache) AllowOpenAIRuntimeBreakerProbe(context.Context, int64, string, string, time.Duration) (bool, error) { + panic("diagnostic must not claim a probe") +} + +func TestCodexQualityHealthReadDoesNotClaimOrClear(t *testing.T) { + cache := &qualityBreakerReadCache{} + s := &OpenAIGatewayService{cache: cache} + account := &Account{ID: 16380, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + expired := time.Now().Add(-time.Minute) + s.openaiAccountRuntimeBlockUntil.Store(account.ID, expired) + state := s.getOpenAIAccountModelTransientState() + key := openAIAccountModelKey{AccountID: account.ID, Model: codexQualityModel} + entry := openAIAccountModelTransientEntry{blockUntil: expired, failureStreak: 2, lastFailure: expired, lastTouched: expired} + state.entries[key] = entry + require.False(t, s.codexQualityHealthBlocked(context.Background(), account)) + value, exists := s.openaiAccountRuntimeBlockUntil.Load(account.ID) + require.True(t, exists) + require.Equal(t, expired, value) + require.Equal(t, entry, state.entries[key]) + require.Equal(t, 1, cache.reads) + cache.blocked = true + require.True(t, s.codexQualityHealthBlocked(context.Background(), account), "ordinary half-open marker stays unavailable to diagnostics") +} + +func TestCodexQualityKeyRevalidationRejectsExpiredQuotaAndDrift(t *testing.T) { + group := int64(53) + key := APIKey{ID: 101, UserID: 9, GroupID: &group, Status: StatusAPIKeyActive} + require.True(t, codexQualityKeyUsable(&key, 9, 101, 53)) + for _, change := range []func(*APIKey){func(k *APIKey) { k.Status = StatusAPIKeyDisabled }, func(k *APIKey) { t := time.Now().Add(-time.Second); k.ExpiresAt = &t }, func(k *APIKey) { k.Quota, k.QuotaUsed = 1, 1 }, func(k *APIKey) { k.UserID++ }, func(k *APIKey) { group := int64(54); k.GroupID = &group }} { + other := key + change(&other) + require.False(t, codexQualityKeyUsable(&other, 9, 101, 53)) + store, run := qualityStateFixture(t, 3) + rt := &codexQualityRuntime{store: store, installation: &PluginInstallation{}} + e := &codexQualityExecution{runtime: rt, runID: run.RunID, grantDigest: run.GrantDigest, accountID: run.AccountID, stage: "acquire", operationID: uuid.NewString(), keyLookup: func(context.Context, int64) (*APIKey, error) { return &other, nil }} + ctx := context.WithValue(context.Background(), codexQualityExecutionKey{}, e) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://chatgpt.com/backend-api/codex/responses", strings.NewReader(`{"model":"gpt-6-astra"}`)) + require.NoError(t, err) + require.ErrorIs(t, reserveCodexQualitySend(req, &Account{ID: run.AccountID}, nil), ErrCodexQualityUnavailable) + saved, _, err := readCodexQualityRun(context.Background(), store, run.RunID) + require.NoError(t, err) + require.Zero(t, saved.UsedSends, "revoked key must fail before the probe budget is spent") + } +} diff --git a/backend/internal/service/codex_routing_host.go b/backend/internal/service/codex_routing_host.go index abfc01f4ed40..606c52b6cef7 100644 --- a/backend/internal/service/codex_routing_host.go +++ b/backend/internal/service/codex_routing_host.go @@ -102,7 +102,7 @@ func (h *pluginExtensionHost) callCodexRouting(ctx context.Context, in extension result.Observation.Code = "routing_budget_spent" return marshal(result) } - clockKey := "clock." + codexRoutingScopeKey(scope) + clockKey := codexQualityClockKey(ctx, "clock."+codexRoutingScopeKey(scope)) clockRecord, err := h.state.ReadExtensionState(ctx, h.key, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: clockKey}) if err != nil { return extensionv1.Result{}, errCodexRoutingUnavailable @@ -307,8 +307,12 @@ func (s *OpenAIGatewayService) ExecuteCodexRoutingProbe(ctx context.Context, que if !ok { return nil, scope, errCodexRoutingUnavailable } + if err := reserveCodexQualitySend(request, a, nil); err != nil { + return nil, scope, err + } response, lease, err := transport.DoWithCodexConnectionLease(request, resolveAccountProxyURL(a), a.ID, codexRoutingScopeKey(scope), "", deadline, nil) scope.ConnectionLeaseID, scope.RouteEvidence = lease, "connection" + observeCodexQualityResponse(ctx, response, err, &extensionv1.CodexRoutingQualification{Scope: scope, Model: query.Model}) if err == nil && response != nil { observeCodexInfrastructureCookies(scope, response.Header, deadline) s.observeCodexWire(ctx, a, request, response, &extensionv1.CodexRoutingQualification{Scope: scope, Model: query.Model}) diff --git a/backend/internal/service/codex_routing_proxy.go b/backend/internal/service/codex_routing_proxy.go index e515c5332af4..3ce55371f18b 100644 --- a/backend/internal/service/codex_routing_proxy.go +++ b/backend/internal/service/codex_routing_proxy.go @@ -44,7 +44,12 @@ func (s *OpenAIGatewayService) doCodexRoutingAcquisition(request *http.Request, if s.httpUpstream == nil { return nil, errCodexRoutingUnavailable } - return s.httpUpstream.Do(wire, normal, accountID, 1) + if err := reserveCodexQualityAcquisition(wire, accountID); err != nil { + return nil, err + } + response, err := s.httpUpstream.Do(wire, normal, accountID, 1) + observeCodexQualityResponse(wire.Context(), response, err, nil) + return response, err } var trust codexRoutingProxyTrust if len(record.Value) > 128*1024 || json.Unmarshal(record.Value, &trust) != nil || len(trust.Certificates) != 1 { @@ -62,7 +67,12 @@ func (s *OpenAIGatewayService) doCodexRoutingAcquisition(request *http.Request, } defer transport.CloseIdleConnections() client := &http.Client{Transport: transport, Timeout: 25 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} - return client.Do(wire) + if err := reserveCodexQualityAcquisition(wire, accountID); err != nil { + return nil, err + } + response, err := client.Do(wire) + observeCodexQualityResponse(wire.Context(), response, err, nil) + return response, err } func verifyCodexRoutingProxyCertificate(state tls.ConnectionState, pinnedRoots *x509.CertPool) error { diff --git a/backend/internal/service/codex_routing_redaction.go b/backend/internal/service/codex_routing_redaction.go index d514e182cf97..e324f91fff57 100644 --- a/backend/internal/service/codex_routing_redaction.go +++ b/backend/internal/service/codex_routing_redaction.go @@ -11,6 +11,9 @@ import ( ) func fixedCodexRoutingBundleKey(ctx context.Context, query extensionv1.CodexRoutingQuery, kind string) string { + if key := codexQualityBundleKey(ctx, kind); key != "" { + return key + } suffix := "" if _, validation := codexValidationFromContext(ctx); validation { suffix = ".validation" diff --git a/backend/internal/service/codex_routing_transport.go b/backend/internal/service/codex_routing_transport.go index 488df1dd32af..a4a84441f543 100644 --- a/backend/internal/service/codex_routing_transport.go +++ b/backend/internal/service/codex_routing_transport.go @@ -45,7 +45,14 @@ func (s *OpenAIGatewayService) prepareQualifiedCodexRequest(request *http.Reques if request.URL == nil || request.URL.Scheme != "https" || request.URL.Hostname() != "chatgpt.com" || (request.URL.Path != "/backend-api/codex/responses" && request.URL.Path != "/backend-api/codex/responses/compact") { return request, nil, nil, errCodexRoutingUnavailable } - q, installation, err := s.pluginManager.codexRoutingQualification(request.Context(), account, model) + var q *extensionv1.CodexRoutingQualification + var installation *PluginInstallation + var err error + if IsCodexQualityRequest(request.Context()) { + q, installation, err = codexQualityRequestQualification(request.Context(), account, model) + } else { + q, installation, err = s.pluginManager.codexRoutingQualification(request.Context(), account, model) + } if err != nil { return request, nil, installation, err } @@ -85,7 +92,17 @@ func (s *OpenAIGatewayService) doQualifiedCodexUpstream(request *http.Request, a return nil, true, errCodexRoutingUnavailable } start := time.Now() + if err := reserveCodexQualitySend(wire, account, q); err != nil { + return nil, true, err + } response, _, err := transport.DoWithCodexConnectionLease(wire, proxyURL, account.ID, codexRoutingScopeKey(q.Scope), q.Scope.ConnectionLeaseID, q.ExpiresAt, nil) + if IsCodexQualityRequest(request.Context()) { + if response != nil { + observeCodexInfrastructureCookies(q.Scope, response.Header, q.ExpiresAt) + } + observeCodexQualityResponse(request.Context(), response, err, q) + return response, true, err + } if err != nil { s.publishCodexRoutingObservation(request.Context(), account, installation, q, extensionv1.CodexRoutingObservation{Code: "routing_connection_expired", Stage: "business", RequestedModel: model, ObservedAt: time.Now().UTC(), Transport: "http", CookieSent: false}, nil) return nil, true, err @@ -257,7 +274,7 @@ func (s *OpenAIGatewayService) recordCodexRoutingDeletions(ctx context.Context, ctx = WithPluginExecution(context.WithoutCancel(ctx), installation) ctx, cancel := context.WithTimeout(ctx, time.Second) defer cancel() - key := "clock." + codexRoutingScopeKey(q.Scope) + key := codexQualityClockKey(ctx, "clock."+codexRoutingScopeKey(q.Scope)) for range 3 { record, err := store.ReadExtensionState(ctx, installation.PluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: key}) var clock codexRoutingCookieClock diff --git a/backend/internal/service/codex_ticket_plugin.go b/backend/internal/service/codex_ticket_plugin.go index 686bff9b2467..696ad726575d 100644 --- a/backend/internal/service/codex_ticket_plugin.go +++ b/backend/internal/service/codex_ticket_plugin.go @@ -51,6 +51,10 @@ func (s *OpenAIGatewayService) openAICodexTicketEnabledContext(context.Context) } func (s *OpenAIGatewayService) applyOpenAICodexTicket(ctx context.Context, account *Account, model string, headers http.Header) error { + if IsCodexQualityRequest(ctx) { + _, _, err := codexQualityRequestQualification(ctx, account, model) + return err + } if s == nil || s.pluginManager == nil { return nil } diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index 11c328888647..4f6f01e56f2b 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -450,6 +450,11 @@ func isOpenAIAccount(account *Account) bool { // handleOpenAIAccountUpstreamError expects canonicalModel to be the model used // for scheduling after applying account mapping exactly once. func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Context, account *Account, statusCode int, headers http.Header, responseBody []byte, canonicalModel ...string) bool { + // A scoped quality experiment records its own failures without changing + // ordinary account health, cooldowns, linked accounts or scheduling state. + if IsCodexQualityRequest(ctx) { + return false + } // Classify request-local failures before any policy can mutate account health. // The same predicates are used by the HTTP failover and error constructors. if isOpenAIRequestScopedSafetyRejection(responseBody) { diff --git a/backend/internal/service/openai_codex_request_compression.go b/backend/internal/service/openai_codex_request_compression.go index d1e885af93c8..4c788b4e2647 100644 --- a/backend/internal/service/openai_codex_request_compression.go +++ b/backend/internal/service/openai_codex_request_compression.go @@ -141,6 +141,9 @@ func (s *OpenAIGatewayService) doOpenAICodexUpstream(req *http.Request, account if response, handled, routingErr := s.doQualifiedCodexUpstream(wire, account, proxyURL); handled { return response, routingErr } + if IsCodexQualityRequest(wire.Context()) { + return nil, ErrCodexQualityUnavailable + } response, err := s.httpUpstream.Do(wire, proxyURL, account.ID, account.Concurrency) if err == nil && isOpenAICodexTicketAccount(account) { s.observeCodexWire(wire.Context(), account, wire, response, nil) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 7f5d2a39d984..d3ddd7457d9f 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -2681,7 +2681,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( ms := int(time.Since(startTime).Milliseconds()) firstTokenMs = &ms } - s.clearOpenAIProxyStreamDisconnect(account) + s.clearOpenAIProxyStreamDisconnect(account, ctx) logger.FromContext(ctx).Info("openai.refusal_recovery_rewritten", zap.String("transport", "sse"), zap.Bool("early", refusalEarlyEmitted)) return resultWithUsage(), nil case openAIRefusalStreamHold: @@ -2758,7 +2758,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( if clientDisconnected { return resultWithUsage(), fmt.Errorf("stream usage incomplete after disconnect: %w", err) } - s.recordOpenAIProxyStreamDisconnect(account, err, upstreamRequestID) + s.recordOpenAIProxyStreamDisconnect(account, err, upstreamRequestID, ctx) logger.LegacyPrintf("service.openai_gateway", "[OpenAI passthrough] 流读取异常中断: account=%d request_id=%s err=%v", account.ID, @@ -2787,14 +2787,14 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, nil, "OpenAI stream ended before a terminal event", resp.StatusCode) } s.recordOpenAIStreamUpstreamError(c, account, true, upstreamRequestID, "stream_error", nil, "OpenAI stream ended before an authoritative terminal event") - s.recordOpenAIProxyStreamDisconnect(account, errors.New("stream ended before terminal event"), upstreamRequestID) + s.recordOpenAIProxyStreamDisconnect(account, errors.New("stream ended before terminal event"), upstreamRequestID, ctx) return resultWithUsage(), errors.New("stream usage incomplete: missing terminal event") } if terminalEventType != "response.completed" && terminalEventType != "response.done" { return resultWithUsage(), fmt.Errorf("upstream response terminated with %s", terminalEventType) } if sawTerminalEvent && !sawFailedEvent { - s.clearOpenAIProxyStreamDisconnect(account) + s.clearOpenAIProxyStreamDisconnect(account, ctx) } logOpenAISuccessMissingUsage(ctx, c, account, resp, usage, terminalEventType, clientDisconnected) diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 03b6dd118492..ae8a97d38ecc 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -522,7 +522,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. terminalEventType = "response.failed" } if sawTerminalEvent && !sawFailedEvent && terminalResponseErr == nil { - s.clearOpenAIProxyStreamDisconnect(account) + s.clearOpenAIProxyStreamDisconnect(account, ctx) } if !sawTerminalEvent && !openAIStreamClientOutputStarted(c, clientOutputStarted) && !eventShouldFlush { return resultWithUsage(), s.newOpenAIStreamFailoverError( @@ -539,7 +539,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if !sawTerminalEvent { s.recordOpenAIStreamUpstreamError(c, account, false, upstreamRequestID, "stream_error", nil, "OpenAI stream ended before an authoritative terminal event") if openAIStreamClientOutputStarted(c, clientOutputStarted) && !clientDisconnected { - s.recordOpenAIProxyStreamDisconnect(account, errors.New("stream ended before terminal event"), upstreamRequestID) + s.recordOpenAIProxyStreamDisconnect(account, errors.New("stream ended before terminal event"), upstreamRequestID, ctx) } return resultWithUsage(), fmt.Errorf("stream usage incomplete: missing terminal event") } @@ -608,7 +608,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if clientDisconnected { return resultWithUsage(), fmt.Errorf("stream usage incomplete after disconnect: %w", scanErr), true } - s.recordOpenAIProxyStreamDisconnect(account, scanErr, upstreamRequestID) + s.recordOpenAIProxyStreamDisconnect(account, scanErr, upstreamRequestID, ctx) code, message := classifyOpenAIUpstreamStreamReadError(scanErr) sendErrorEvent(code, message) return resultWithUsage(), fmt.Errorf("stream read error: %w", scanErr), true @@ -1641,6 +1641,9 @@ func extractOpenAIResponseIDFromJSONBytes(body []byte) string { } func (s *OpenAIGatewayService) bindHTTPResponseAccount(ctx context.Context, c *gin.Context, account *Account, responseID string) { + if IsCodexQualityRequest(ctx) { + return + } if s == nil || account == nil || account.ID <= 0 { return } diff --git a/backend/internal/service/openai_gateway_usage.go b/backend/internal/service/openai_gateway_usage.go index 390043e08bd3..a07c6d367e39 100644 --- a/backend/internal/service/openai_gateway_usage.go +++ b/backend/internal/service/openai_gateway_usage.go @@ -20,6 +20,9 @@ import ( // OpenAIRecordUsageInput input for recording usage type OpenAIRecordUsageInput struct { + // Captured from the authenticated diagnostic context before async billing. + // Accounting is unchanged; diagnostic success cannot recover account health. + CodexQuality bool Result *OpenAIForwardResult APIKey *APIKey User *User @@ -154,13 +157,7 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec if result == nil { return errors.New("openai usage result is nil") } - if !input.CyberBlocked && input.Account != nil && s.cindyHealth != nil && - hasCanonicalCindyProviderIdentity(input.Account) { - s.cindyHealth.ObserveCindyHealthSuccess(ctx, input.Account) - } - if s.rateLimitService != nil && input.Account != nil && input.Account.Platform == PlatformOpenAI { - s.rateLimitService.ResetOpenAI403Counter(ctx, input.Account.ID) - } + s.observeOpenAIUsageAccountHealth(ctx, input) apiKey := input.APIKey user := input.User @@ -1314,15 +1311,28 @@ func (s *OpenAIGatewayService) updateCodexUsageSnapshot(ctx context.Context, acc return } + qualityObservation := IsCodexQualityRequest(ctx) go func() { updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err == nil { + if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err == nil && !qualityObservation { notifyOpenAIAutoReset(accountID) } }() } +func (s *OpenAIGatewayService) observeOpenAIUsageAccountHealth(ctx context.Context, input *OpenAIRecordUsageInput) { + if input.CodexQuality { + return + } + if !input.CyberBlocked && input.Account != nil && s.cindyHealth != nil && hasCanonicalCindyProviderIdentity(input.Account) { + s.cindyHealth.ObserveCindyHealthSuccess(ctx, input.Account) + } + if s.rateLimitService != nil && input.Account != nil && input.Account.Platform == PlatformOpenAI { + s.rateLimitService.ResetOpenAI403Counter(ctx, input.Account.ID) + } +} + func (s *OpenAIGatewayService) UpdateCodexUsageSnapshotFromHeaders(ctx context.Context, accountID int64, headers http.Header) { if accountID <= 0 || headers == nil { return diff --git a/backend/internal/service/openai_proxy_stream_circuit.go b/backend/internal/service/openai_proxy_stream_circuit.go index 575fc134bb76..cd8efac91d34 100644 --- a/backend/internal/service/openai_proxy_stream_circuit.go +++ b/backend/internal/service/openai_proxy_stream_circuit.go @@ -235,7 +235,10 @@ func openAIProxyStreamCircuitProxyID(account *Account) (int64, bool) { return *account.ProxyID, true } -func (s *OpenAIGatewayService) recordOpenAIProxyStreamDisconnect(account *Account, streamErr error, upstreamRequestID string) { +func (s *OpenAIGatewayService) recordOpenAIProxyStreamDisconnect(account *Account, streamErr error, upstreamRequestID string, contexts ...context.Context) { + if len(contexts) > 0 && IsCodexQualityRequest(contexts[0]) { + return + } proxyID, ok := openAIProxyStreamCircuitProxyID(account) if !ok || streamErr == nil || errors.Is(streamErr, context.Canceled) || errors.Is(streamErr, context.DeadlineExceeded) { return @@ -255,7 +258,10 @@ func (s *OpenAIGatewayService) recordOpenAIProxyStreamDisconnect(account *Accoun ) } -func (s *OpenAIGatewayService) clearOpenAIProxyStreamDisconnect(account *Account) { +func (s *OpenAIGatewayService) clearOpenAIProxyStreamDisconnect(account *Account, contexts ...context.Context) { + if len(contexts) > 0 && IsCodexQualityRequest(contexts[0]) { + return + } proxyID, ok := openAIProxyStreamCircuitProxyID(account) if !ok { return diff --git a/backend/internal/service/openai_upstream_transport_error.go b/backend/internal/service/openai_upstream_transport_error.go index 7af618aba522..02045e7e7978 100644 --- a/backend/internal/service/openai_upstream_transport_error.go +++ b/backend/internal/service/openai_upstream_transport_error.go @@ -106,6 +106,9 @@ func classifyUpstreamTransportError(err error) upstreamTransportErrorClass { // // passthrough tags the Ops error event for the OpenAI passthrough forward path. func (s *OpenAIGatewayService) handleOpenAIUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error, passthrough bool) error { + if IsCodexQualityRequest(ctx) { + return err + } safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 5f0b25b4cfa0..eb6ce38e9c4e 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -2846,6 +2846,9 @@ func truncateTempUnschedMessage(body []byte, maxBytes int) string { // 根据系统设置决定是否标记账户为临时不可调度或错误状态 // 返回是否应该停止该账号的调度 func (s *RateLimitService) HandleStreamTimeout(ctx context.Context, account *Account, model string) bool { + if IsCodexQualityRequest(ctx) { + return false + } if account == nil { return false } diff --git a/plugins/account-tools/cmd/plugin/main.go b/plugins/account-tools/cmd/plugin/main.go index 5002f1f43303..418923eacd1c 100644 --- a/plugins/account-tools/cmd/plugin/main.go +++ b/plugins/account-tools/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.account-tools", Version: version, Capabilities: []string{extensionv1.CapabilityAdmin, extensionv1.CapabilityUI}}, policy.New()) diff --git a/plugins/account-tools/manifest.source.json b/plugins/account-tools/manifest.source.json index 5e1b7bd83305..4ffabca969cd 100644 --- a/plugins/account-tools/manifest.source.json +++ b/plugins/account-tools/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.account-tools", "name": "账号工具", - "version": "0.2.10", + "version": "0.2.11", "description": "账号分类操作策略与连接测试选项", "author": "HTExplicit", "requires": { "sub2api": ">=0.2.7-codexrip.2 <0.2.9", "tested_sub2api_versions": [], "plugin_protocol": 1, "transport_api": 1, "ui_bridge": 1, "extension_api": 1 }, diff --git a/plugins/admin-observability/cmd/plugin/main.go b/plugins/admin-observability/cmd/plugin/main.go index ac8c85cc75b3..7a0c22daed00 100644 --- a/plugins/admin-observability/cmd/plugin/main.go +++ b/plugins/admin-observability/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.admin-observability", Version: version, Capabilities: []string{extensionv1.CapabilityObservability, extensionv1.CapabilityAdmin, extensionv1.CapabilityUI}}, policy.New()) diff --git a/plugins/admin-observability/manifest.source.json b/plugins/admin-observability/manifest.source.json index 503744ce49f0..d97350fda0b1 100644 --- a/plugins/admin-observability/manifest.source.json +++ b/plugins/admin-observability/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.admin-observability", "name": "界面与观测", - "version": "0.2.10", + "version": "0.2.11", "description": "账号流量观测规则与指标展示", "author": "HTExplicit", "requires": { diff --git a/plugins/cindy-provider/cmd/plugin/main.go b/plugins/cindy-provider/cmd/plugin/main.go index 658477727abd..4e9122163462 100644 --- a/plugins/cindy-provider/cmd/plugin/main.go +++ b/plugins/cindy-provider/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.cindy-provider", Version: version, Capabilities: []string{extensionv1.CapabilityProvider, extensionv1.CapabilityAdmin}}, catalog.New()) diff --git a/plugins/cindy-provider/manifest.source.json b/plugins/cindy-provider/manifest.source.json index e2dc4cb0adc4..bc0355450730 100644 --- a/plugins/cindy-provider/manifest.source.json +++ b/plugins/cindy-provider/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.cindy-provider", "name": "Cindy Provider", - "version": "0.2.10", + "version": "0.2.11", "description": "Cindy 模型目录、协议能力、价格参考及账号健康判定", "author": "HTExplicit", "requires": { diff --git a/plugins/codex-runtime/cmd/plugin/main.go b/plugins/codex-runtime/cmd/plugin/main.go index 84a3ddde2a1d..6b425023ce1e 100644 --- a/plugins/codex-runtime/cmd/plugin/main.go +++ b/plugins/codex-runtime/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.codex-runtime", Version: version, diff --git a/plugins/codex-runtime/manifest.source.json b/plugins/codex-runtime/manifest.source.json index b8e1eb8eeb52..a99bbda67bb2 100644 --- a/plugins/codex-runtime/manifest.source.json +++ b/plugins/codex-runtime/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.codex-runtime", "name": "Codex 运行扩展", - "version": "0.2.10", + "version": "0.2.11", "description": "Cookie 路由资格、按需有界续获与出站指纹", "author": "HTExplicit", "requires": { diff --git a/plugins/image-tools/cmd/plugin/main.go b/plugins/image-tools/cmd/plugin/main.go index 4f73fa5b0419..7f5003bc6951 100644 --- a/plugins/image-tools/cmd/plugin/main.go +++ b/plugins/image-tools/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.image-tools", Version: version, Capabilities: []string{extensionv1.CapabilityRequest, extensionv1.CapabilityAdmin}}, policy.New()) diff --git a/plugins/image-tools/manifest.source.json b/plugins/image-tools/manifest.source.json index 07231fee152b..027ec34966e8 100644 --- a/plugins/image-tools/manifest.source.json +++ b/plugins/image-tools/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.image-tools", "name": "图像工具", - "version": "0.2.10", + "version": "0.2.11", "description": "Image Studio 请求计划、模型能力和 Responses 图像桥接策略", "author": "HTExplicit", "requires": { "sub2api": ">=0.2.7-codexrip.2 <0.2.9", "tested_sub2api_versions": [], "plugin_protocol": 1, "transport_api": 1, "ui_bridge": 1, "extension_api": 1 }, diff --git a/plugins/model-policy/cmd/plugin/main.go b/plugins/model-policy/cmd/plugin/main.go index bd9f4b928436..03b580a33cad 100644 --- a/plugins/model-policy/cmd/plugin/main.go +++ b/plugins/model-policy/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.model-policy", Version: version, Capabilities: []string{extensionv1.CapabilityCatalog, extensionv1.CapabilityAdmin}}, catalog.New()) diff --git a/plugins/model-policy/manifest.source.json b/plugins/model-policy/manifest.source.json index 1078cabbc1d3..d3c1416413b2 100644 --- a/plugins/model-policy/manifest.source.json +++ b/plugins/model-policy/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.model-policy", "name": "模型目录与容量策略", - "version": "0.2.10", + "version": "0.2.11", "description": "精确模型别名、供应商范围与经过核对的容量参考", "author": "HTExplicit", "requires": { diff --git a/plugins/prompt-skills/cmd/plugin/main.go b/plugins/prompt-skills/cmd/plugin/main.go index ee574a55c37f..4af4fbf8bf74 100644 --- a/plugins/prompt-skills/cmd/plugin/main.go +++ b/plugins/prompt-skills/cmd/plugin/main.go @@ -5,7 +5,7 @@ import ( extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1" ) -var version = "0.2.10" +var version = "0.2.11" func main() { extensionv1.Serve(extensionv1.Definition{ID: "codexrip.prompt-skills", Version: version, Capabilities: []string{extensionv1.CapabilityRequest, extensionv1.CapabilityAdmin}}, policy.New()) diff --git a/plugins/prompt-skills/manifest.source.json b/plugins/prompt-skills/manifest.source.json index 0b0539f49108..d72f477d8e41 100644 --- a/plugins/prompt-skills/manifest.source.json +++ b/plugins/prompt-skills/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.prompt-skills", "name": "提示词与技能策略", - "version": "0.2.10", + "version": "0.2.11", "description": "提示词完整性、应用范围与协议载体策略", "author": "HTExplicit", "requires": { "sub2api": ">=0.2.7-codexrip.2 <0.2.9", "tested_sub2api_versions": [], "plugin_protocol": 1, "transport_api": 1, "ui_bridge": 1, "extension_api": 1 },