From ff891a57db2d5b23c142df1f375272fe9c6dd2e9 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 5 Oct 2026 13:44:13 +0800 Subject: [PATCH] feat(reasoning): hand a request to another account when this one cannot read its ciphertext Ciphertext in a Codex conversation (reasoning items, inter-agent message bodies, encrypted tool arguments) is readable only by the upstream organisation that issued it. When an account of another organisation rejected it and the stripped retry could not help, the request ended on that account and every client retry hit the same account, although another account of the group might accept the request unchanged. StopError now decides once whether an attempt ended as an account mismatch: the upstream said it could not decrypt or verify ciphertext in the request, nothing on this account repaired it, and nothing of the attempt reached the client. The attempt's own failure is returned marked for the next account, in request scope and without a health penalty, instead of as a terminal. An ordinary validation rejection, a request with server-held history, an account status, a cancelled request and the switched-off recovery never are. The Responses loop hands a mismatch to the next account without the same-account retry policy, at most three times per request and within the normal switch budget. When accounts run out the client receives the first mismatch failure, as before; an account that ends the request with its own answer is still the answer. A sticky binding this request moved goes back to the first rejecting account when no account answered, and stays with an account that accepted or was already answering; a binding the scheduler kept elsewhere is not touched. The continuation diagnostic marks such an attempt with recovery.account_mismatch. Chat Completions, Messages and WebSocket are unchanged. Co-Authored-By: Claude Opus 5.5 --- .downstream/gpt6-models-and-prompts.md | 2 +- ...enai_continuation_account_mismatch_test.go | 279 ++++++++++++++++++ .../handler/openai_gateway_handler.go | 106 ++++++- backend/internal/service/gateway_service.go | 1 + .../service/openai_continuation_diagnostic.go | 4 + .../service/openai_gateway_forward.go | 7 +- .../service/openai_gateway_scheduling.go | 20 ++ .../openai_gateway_service_hotpath_test.go | 4 +- .../service/openai_gateway_upstream_errors.go | 67 ++++- .../service/openai_oauth_passthrough_test.go | 25 +- .../service/openai_reasoning_recovery.go | 63 +++- ...openai_reasoning_recovery_terminal_test.go | 21 +- .../service/openai_reasoning_recovery_test.go | 73 +++++ .../service/openai_sticky_compat_test.go | 19 ++ .../openai_ws_protocol_forward_test.go | 6 +- 15 files changed, 650 insertions(+), 47 deletions(-) create mode 100644 backend/internal/handler/openai_continuation_account_mismatch_test.go diff --git a/.downstream/gpt6-models-and-prompts.md b/.downstream/gpt6-models-and-prompts.md index 106449e1afb8..7c9366103fbf 100644 --- a/.downstream/gpt6-models-and-prompts.md +++ b/.downstream/gpt6-models-and-prompts.md @@ -66,7 +66,7 @@ Claude OAuth 伪装系统块只由上游设置决定：设置 → 网关的 `ena 侧栏“扩展功能 → 推理恢复”（`/admin/reasoning-recovery`）：一个全局开关和保存按钮，默认开启，没有账号级选项；页面打开时读取一次，读取失败时显示错误、不显示开关。存为 `settings.reasoning_recovery_config` 一行 JSON（`{"enabled": }`），没有该行即开启；`GET/PUT /api/v1/admin/reasoning-recovery` 以标准响应封装读写 `{"enabled": }`，管理员认证、无二次验证，保存记入操作审计，请求体不是严格的 `{"enabled": }` 时返回 400 且不保存。请求路径只读内存中的布尔值（保存即替换，60 秒后台刷新以同步多实例），不查询数据库，每次转发开始时读取一次。启动时读取失败或存储值不是严格的 `{"enabled": }` 则停止启动；运行中刷新失败保留当前值。 -开启时，OpenAI 平台的 API Key、OAuth 和 Setup Token 账号（含自动透传）在原生 Responses（含 compact）与 Chat Completions 转 Responses 的 HTTP 请求上生效：上游以结构化错误码 `invalid_encrypted_content` 或 `thinking_signature_invalid` 拒绝、且尚无语义输出时，去掉推理项的 `encrypted_content`，在同一账号重发一次：上游用 `input[N]` 指明条目时只去掉该项，否则去掉请求里全部推理项的密文。上游未用 `input[N]` 指明条目、而请求里还有其他密文载体（压缩项、多智能体 `agent_message` 的正文段）时，只有上游错误消息按条目 id 点名的是带密文的推理项才恢复，否则不恢复；这条点名规则只在推理项 id 随请求发出的路径上成立（API Key 账号与自动透传），OAuth / Setup Token 的非透传请求发出前会去掉推理项 id，遇到其他密文载体仍不恢复。其他载体的密文任何情况下都不改动。被去掉的密文按来源在 Redis 记忆 24 小时，期间同源请求发送前即去掉；记忆按每批 64 条读写，读与写共用每个请求 100 毫秒的 I/O 预算。Messages 桥接与 WebSocket 不恢复。关闭时不重发也不预先去除，签名拒绝仍作为请求级终态返回，不切换账号。 +开启时，OpenAI 平台的 API Key、OAuth 和 Setup Token 账号（含自动透传）在原生 Responses（含 compact）与 Chat Completions 转 Responses 的 HTTP 请求上生效：上游以结构化错误码 `invalid_encrypted_content` 或 `thinking_signature_invalid` 拒绝、且尚无语义输出时，去掉推理项的 `encrypted_content`，在同一账号重发一次：上游用 `input[N]` 指明条目时只去掉该项，否则去掉请求里全部推理项的密文。上游未用 `input[N]` 指明条目、而请求里还有其他密文载体（压缩项、多智能体 `agent_message` 的正文段）时，只有上游错误消息按条目 id 点名的是带密文的推理项才恢复，否则不恢复；这条点名规则只在推理项 id 随请求发出的路径上成立（API Key 账号与自动透传），OAuth / Setup Token 的非透传请求发出前会去掉推理项 id，遇到其他密文载体仍不恢复。其他载体的密文任何情况下都不改动。被去掉的密文按来源在 Redis 记忆 24 小时，期间同源请求发送前即去掉；记忆按每批 64 条读写，读与写共用每个请求 100 毫秒的 I/O 预算。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效、恢复重发遇到 5xx）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出这样结束的尝试。Messages 桥接与 WebSocket 不恢复。关闭时不重发也不预先去除，签名拒绝仍作为请求级终态返回，不切换账号。 迁移 `267_purge_account_reasoning_options.sql` 一次性删除各账号 `extra` 中原有的两个账号级推理选项键（键名见该文件）；此后没有代码读写它们，已存的账号级取值不带入全局开关。Chat Completions 转 Responses 的请求只由客户端消息转换而来，网关不保存响应中的推理项。 diff --git a/backend/internal/handler/openai_continuation_account_mismatch_test.go b/backend/internal/handler/openai_continuation_account_mismatch_test.go new file mode 100644 index 000000000000..5f751a42c463 --- /dev/null +++ b/backend/internal/handler/openai_continuation_account_mismatch_test.go @@ -0,0 +1,279 @@ +//go:build unit + +package handler + +import ( + "io" + "net/http" + "strings" + "sync" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +// continuationMismatchUpstream answers by account: an accepting account +// completes the request, every other account answers with its scripted reply +// or, by default, rejects the ciphertext in the request. +type continuationMismatchUpstream struct { + service.HTTPUpstream + mu sync.Mutex + accountIDs []int64 + bodies [][]byte + accepting map[int64]bool + replies map[int64]continuationMismatchReply + rejection continuationMismatchReply +} + +type continuationMismatchReply struct { + status int + payload string + // stream, when set, is sent as an event stream in place of payload. + stream string +} + +func (u *continuationMismatchUpstream) Do(req *http.Request, _ string, accountID int64, _ int) (*http.Response, error) { + body, err := io.ReadAll(req.Body) + if err != nil { + return nil, err + } + u.mu.Lock() + u.accountIDs = append(u.accountIDs, accountID) + u.bodies = append(u.bodies, body) + u.mu.Unlock() + if u.accepting[accountID] { + return openAIOfficialHTTPSuccess(openAIOfficialHTTPAttempt{accountID: accountID, path: req.URL.Path, body: body}), nil + } + reply, scripted := u.replies[accountID] + if !scripted { + reply = u.rejection + } + if reply.stream != "" { + return openAIOfficialHTTPResponse(reply.status, "text/event-stream", reply.stream), nil + } + return openAIOfficialHTTPResponse(reply.status, "application/json", reply.payload), nil +} + +func (u *continuationMismatchUpstream) distinctAccounts() []int64 { + u.mu.Lock() + defer u.mu.Unlock() + var out []int64 + seen := map[int64]bool{} + for _, id := range u.accountIDs { + if !seen[id] { + seen[id] = true + out = append(out, id) + } + } + return out +} + +var ( + // No item is named: with another ciphertext carrier nothing is removed. + continuationMismatchUnrecoverable = continuationMismatchReply{status: http.StatusBadRequest, + payload: `{"error":{"message":"invalid encrypted content","type":"invalid_request_error","param":null,"code":"invalid_encrypted_content"}}`} + // The rejection names the reasoning item, so its ciphertext is removed and + // the request is sent once more. + continuationMismatchNamedReasoning = continuationMismatchReply{status: http.StatusBadRequest, + payload: `{"error":{"message":"The encrypted content for item rs_a could not be verified.","type":"invalid_request_error","param":null,"code":"invalid_encrypted_content"}}`} + // The wording for an inter-agent message body, which carries no code. + continuationMismatchUndecryptable = continuationMismatchReply{status: http.StatusBadRequest, + payload: `{"error":{"type":"invalid_request_error","message":"Encrypted function output content could not be decrypted or decoded."}}`} +) + +func continuationMismatchAccounts(count int, accountType string) []service.Account { + accounts := make([]service.Account, 0, count) + for id := int64(1); id <= int64(count); id++ { + account := service.Account{ + ID: id, Name: "organisation", Platform: service.PlatformOpenAI, Type: accountType, + Status: service.StatusActive, Schedulable: true, Priority: int(id), GroupIDs: []int64{openAIOfficialHTTPGroupID}, + Credentials: map[string]any{"api_key": "fixture", "base_url": "https://api.example.test"}, + Extra: map[string]any{"use_responses_api": true}, + } + if accountType == service.AccountTypeOAuth { + account.Credentials = map[string]any{"access_token": "fixture"} + } + accounts = append(accounts, account) + } + return accounts +} + +// newContinuationMismatchHandler runs the real handler, selector and forwarder +// over in-memory accounts and a sticky-session cache the test can read. +func newContinuationMismatchHandler(t *testing.T, upstream service.HTTPUpstream, accounts []service.Account) (*OpenAIGatewayHandler, *openAIOfficialHTTPStickyCache) { + t.Helper() + repo := &openAIOfficialHTTPAccountRepo{grokCredentialHandlerRepo: grokCredentialHandlerRepo{accounts: accounts}} + cache := &openAIOfficialHTTPStickyCache{grokCredentialHandlerGatewayCache: grokCredentialHandlerGatewayCache{ + sessions: make(map[grokCredentialHandlerGatewayCacheKey]int64), + }} + cfg := &config.Config{RunMode: config.RunModeSimple} + cfg.Gateway.MaxAccountSwitches = 10 + billing := service.NewBillingCacheService(nil, nil, nil, nil, nil, nil, cfg, nil) + t.Cleanup(billing.Stop) + gateway := service.NewOpenAIGatewayService( + repo, nil, nil, nil, nil, nil, cache, cfg, nil, nil, + service.NewBillingService(cfg, nil), service.NewRateLimitService(repo, nil, cfg, nil, nil), billing, upstream, + &service.DeferredService{}, nil, nil, nil, nil, nil, nil, nil, + ) + h := NewOpenAIGatewayHandler(gateway, service.NewConcurrencyService(nil), billing, + service.NewAPIKeyService(nil, nil, nil, nil, nil, nil, cfg), nil, nil, nil, nil, cfg) + return h, cache +} + +func serveContinuationMismatch(t *testing.T, h *OpenAIGatewayHandler, body string) *responseRecorderResult { + t.Helper() + c, recorder := newReplayableEndpointContext(t, openAIOfficialHTTPGroupID, http.MethodPost, "/v1/responses", []byte(body)) + c.Request.Header.Set("session_id", "one-conversation") + h.Responses(c) + return &responseRecorderResult{code: recorder.Code, body: recorder.Body.String()} +} + +type responseRecorderResult struct { + code int + body string +} + +func continuationMismatchBoundAccount(t *testing.T, cache *openAIOfficialHTTPStickyCache) int64 { + t.Helper() + // The cache also holds response-owner entries; the session binding is the + // entry keyed by the session hash alone. + var bound []int64 + for key, accountID := range cache.sessions { + if strings.Count(key.sessionHash, ":") == 1 { + bound = append(bound, accountID) + } + } + require.Len(t, bound, 1, "one conversation has one binding") + return bound[0] +} + +// A Codex multi-agent history: reasoning ciphertext plus an inter-agent message +// whose body is ciphertext the gateway cannot remove. +const continuationMismatchBody = `{"model":"gpt-5.1","stream":false,"input":[{"type":"reasoning","id":"rs_a","encrypted_content":"cipher-reasoning","summary":[]},{"type":"agent_message","id":"amsg_b","author":"/root","recipient":"/root/worker","content":[{"type":"input_text","text":"Payload:"},{"type":"encrypted_content","encrypted_content":"cipher-agent"}]}]}` + +func TestOpenAIGatewayHandler_ContinuationCiphertextMismatchMovesToAnAccountThatAccepts(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, test := range []struct { + name string + rejection continuationMismatchReply + callsOnFirst int + }{ + // The stripped retry is rejected again, as when the message body is foreign too. + {"stripped_retry_rejected", continuationMismatchNamedReasoning, 2}, + {"recovery_not_possible", continuationMismatchUnrecoverable, 1}, + } { + t.Run(test.name, func(t *testing.T) { + upstream := &continuationMismatchUpstream{accepting: map[int64]bool{2: true}, rejection: test.rejection} + handler, cache := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(2, service.AccountTypeAPIKey)) + response := serveContinuationMismatch(t, handler, continuationMismatchBody) + + require.Equal(t, http.StatusOK, response.code, response.body) + require.Contains(t, response.body, "healthy answer") + require.NotContains(t, response.body, service.OpenAIContinuationStateUnavailableCode) + require.Len(t, upstream.accountIDs, test.callsOnFirst+1) + require.Equal(t, []int64{1, 2}, upstream.distinctAccounts()) + accepted := upstream.bodies[len(upstream.bodies)-1] + for _, path := range []string{"input.0.encrypted_content", "input.1"} { + require.Equal(t, gjson.GetBytes(upstream.bodies[0], path).Raw, gjson.GetBytes(accepted, path).Raw, "the next account receives the request unchanged: "+path) + } + require.Equal(t, "cipher-reasoning", gjson.GetBytes(accepted, "input.0.encrypted_content").String()) + require.Equal(t, int64(2), continuationMismatchBoundAccount(t, cache), "the conversation continues on the account that accepted it") + }) + } +} + +func TestOpenAIGatewayHandler_ContinuationCiphertextMismatchEndsWhenNoAccountAccepts(t *testing.T) { + gin.SetMode(gin.TestMode) + upstream := &continuationMismatchUpstream{rejection: continuationMismatchUnrecoverable} + handler, cache := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(maxOpenAICiphertextAccountSwitches+3, service.AccountTypeAPIKey)) + response := serveContinuationMismatch(t, handler, continuationMismatchBody) + + require.Len(t, upstream.distinctAccounts(), maxOpenAICiphertextAccountSwitches+1, "the first account and a bounded number of others") + require.Len(t, upstream.accountIDs, maxOpenAICiphertextAccountSwitches+1, "an account that cannot repair the request is asked once") + require.Equal(t, http.StatusBadRequest, response.code) + require.Equal(t, service.OpenAIContinuationStateUnavailableCode, gjson.Get(response.body, "error.code").String()) + require.NotContains(t, response.body, "invalid encrypted content", "the client receives the fixed sentence, not upstream text") + require.Equal(t, upstream.accountIDs[0], continuationMismatchBoundAccount(t, cache), "the accounts that rejected the conversation are no home for it") +} + +// An account that reads the ciphertext and rejects the request for another +// reason has answered: that answer ends the request. +func TestOpenAIGatewayHandler_ContinuationCiphertextMismatchKeepsALaterAccountsOwnAnswer(t *testing.T) { + gin.SetMode(gin.TestMode) + upstream := &continuationMismatchUpstream{ + rejection: continuationMismatchUnrecoverable, + replies: map[int64]continuationMismatchReply{2: {status: http.StatusBadRequest, + payload: `{"error":{"type":"invalid_request_error","code":"missing_required_parameter","param":"input[3].call_id","message":"do-not-leak-validation"}}`}}, + } + handler, cache := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(3, service.AccountTypeAPIKey)) + response := serveContinuationMismatch(t, handler, continuationMismatchBody) + + require.Equal(t, []int64{1, 2}, upstream.accountIDs, "a validation rejection is the same on every account") + require.Equal(t, http.StatusBadRequest, response.code) + require.Equal(t, service.OpenAIRequestRejectedCode, gjson.Get(response.body, "error.code").String()) + require.NotContains(t, response.body, "do-not-leak") + require.Equal(t, int64(1), continuationMismatchBoundAccount(t, cache)) +} + +// A ciphertext rejection wrapped in a 5xx is still no fault of the account: +// the account is not asked again before the next one is tried. +func TestOpenAIGatewayHandler_ContinuationCiphertextMismatchSkipsTheSameAccountRetry(t *testing.T) { + gin.SetMode(gin.TestMode) + upstream := &continuationMismatchUpstream{ + accepting: map[int64]bool{2: true}, + rejection: continuationMismatchReply{status: http.StatusServiceUnavailable, payload: `{"error":{"message":"Encrypted content could not be verified"}}`}, + } + handler, _ := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(2, service.AccountTypeOAuth)) + response := serveContinuationMismatch(t, handler, strings.Replace(continuationMismatchBody, `"stream":false`, `"stream":true`, 1)) + + require.Equal(t, []int64{1, 2}, upstream.accountIDs) + require.Contains(t, response.body, "response.completed") +} + +// When the accounts run out, the client receives what the first account +// answered, in that failure's own classification, and the session is where it +// was before the others were tried. +func TestOpenAIGatewayHandler_ContinuationCiphertextMismatchRendersTheFirstFailureWhenAccountsRunOut(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, test := range []struct { + name string + first, second continuationMismatchReply + code string + }{ + {"continuation_state_first", continuationMismatchUnrecoverable, continuationMismatchUndecryptable, service.OpenAIContinuationStateUnavailableCode}, + {"request_rejected_first", continuationMismatchUndecryptable, continuationMismatchUnrecoverable, service.OpenAIRequestRejectedCode}, + } { + t.Run(test.name, func(t *testing.T) { + upstream := &continuationMismatchUpstream{replies: map[int64]continuationMismatchReply{1: test.first, 2: test.second}} + handler, cache := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(2, service.AccountTypeAPIKey)) + response := serveContinuationMismatch(t, handler, continuationMismatchBody) + + require.Equal(t, []int64{1, 2}, upstream.accountIDs) + require.Equal(t, http.StatusBadRequest, response.code) + require.Equal(t, test.code, gjson.Get(response.body, "error.code").String()) + require.NotContains(t, response.body, "could not be decrypted") + require.Equal(t, int64(1), continuationMismatchBoundAccount(t, cache)) + }) + } +} + +// An account that read the ciphertext and was answering when its stream broke +// is where the conversation runs: the session stays with it. +func TestOpenAIGatewayHandler_ContinuationCiphertextMismatchKeepsTheSessionOnAnAccountThatWasAnswering(t *testing.T) { + gin.SetMode(gin.TestMode) + upstream := &continuationMismatchUpstream{ + rejection: continuationMismatchUnrecoverable, + replies: map[int64]continuationMismatchReply{2: {status: http.StatusOK, stream: "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_partial\",\"status\":\"in_progress\"}}\n\n" + + "event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"partial-answer\"}\n\n"}}, + } + handler, cache := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(3, service.AccountTypeAPIKey)) + response := serveContinuationMismatch(t, handler, strings.Replace(continuationMismatchBody, `"stream":false`, `"stream":true`, 1)) + + require.Equal(t, []int64{1, 2}, upstream.accountIDs) + require.Contains(t, response.body, "partial-answer") + require.Equal(t, int64(2), continuationMismatchBoundAccount(t, cache)) +} diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 4a87de6d2432..cf99bc70fc10 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -224,6 +224,13 @@ type grokMediaEligibilityProber interface { const maxOpenAIFirstOutputTimeoutSwitches = 1 +// maxOpenAICiphertextAccountSwitches bounds how often one request is handed +// to another account because an account could not read its ciphertext, within +// the ordinary switch budget. Every rejection costs an upload of the whole +// history, and when the account that issued the ciphertext is gone no account +// accepts. +const maxOpenAICiphertextAccountSwitches = 3 + func openAIForwardSucceededForScheduling(result *service.OpenAIForwardResult) bool { return result.SucceededForScheduling() } @@ -723,6 +730,36 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { var passthroughFailoverState openAIPassthroughFailoverState var accountTypePreference service.OpenAIAccountTypePreference var sameAccountRetrySelection *service.AccountSelectionResult + // An account that cannot read the ciphertext in the request hands it to + // the next account. When the accounts or the switch budget run out, the + // client receives the first such failure, as it did before accounts were + // switched for it; an account that ends the request with an answer of its + // own is still the answer. + var ciphertextFailoverErr *service.UpstreamFailoverError + ciphertextFirstAccountID, ciphertextSessionHash := int64(0), "" + ciphertextLaterAccounts := make(map[int64]struct{}) + ciphertextSwitchCount := 0 + ciphertextAccepted := false + exhaustedFailoverErr := func(current *service.UpstreamFailoverError) *service.UpstreamFailoverError { + if ciphertextFailoverErr != nil { + return ciphertextFailoverErr + } + return current + } + // Selecting one of the later accounts may have moved the session to it. + // When none of them answered, they are no better home for the conversation + // than the account that first rejected it, so a binding this request moved + // goes back there. A binding the scheduler kept elsewhere is not touched. + defer func() { + if ciphertextFailoverErr == nil || ciphertextAccepted { + return + } + restoreCtx, cancel := context.WithTimeout(context.WithoutCancel(c.Request.Context()), 2*time.Second) + defer cancel() + if err := h.gatewayService.MoveStickySessionIfBoundTo(restoreCtx, apiKey.GroupID, ciphertextSessionHash, ciphertextFirstAccountID, ciphertextLaterAccounts); err != nil { + reqLog.Warn("openai.ciphertext_mismatch_session_restore_failed", zap.Int64("account_id", ciphertextFirstAccountID), zap.Error(err)) + } + }() // 分组利润控制：请求级装配定价上下文——pricingAt 固定本请求的 // D 与计费高峰因子，选号、槽位终检与全部 failover 重入共用同一门与阈值。 @@ -782,7 +819,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { zap.Int("excluded_account_count", len(failedAccountIDs)), ) if retryingSameAccount && lastFailoverErr != nil { - h.handleFailoverExhausted(c, lastFailoverErr, streamStarted) + h.handleFailoverExhausted(c, exhaustedFailoverErr(lastFailoverErr), streamStarted) return } if len(failedAccountIDs) == 0 { @@ -799,12 +836,16 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { return } if lastFailoverErr != nil { - h.handleFailoverExhausted(c, lastFailoverErr, streamStarted) + h.handleFailoverExhausted(c, exhaustedFailoverErr(lastFailoverErr), streamStarted) } else { h.handleFailoverExhaustedSimple(c, 502, streamStarted) } return } + if (selection == nil || selection.Account == nil) && ciphertextFailoverErr != nil { + h.handleFailoverExhausted(c, ciphertextFailoverErr, streamStarted) + return + } if selection == nil || selection.Account == nil { cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, routingModel, reqModel, requestPlatform) if !cls.ModelNotFound { @@ -831,6 +872,9 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { sessionHash = ensureOpenAIPoolModeSessionHash(sessionHash, account) reqLog.Debug("openai.account_selected", zap.Int64("account_id", account.ID), zap.String("account_name", account.Name)) setOpsSelectedAccount(c, account.ID, account.Platform) + if ciphertextFailoverErr != nil && account.ID != ciphertextFirstAccountID { + ciphertextLaterAccounts[account.ID] = struct{}{} + } var accountReleaseFunc func() var slotResult openAISlotAcquireResult @@ -853,7 +897,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { c, account, account.GetMappedModel(routingModel), lastFailoverErr, failedAccountIDs, &switchCount, maxAccountSwitches, &oauth429FailoverState, "responses", reqLog, true, - func() { h.handleFailoverExhausted(c, lastFailoverErr, streamStarted) }, + func() { h.handleFailoverExhausted(c, exhaustedFailoverErr(lastFailoverErr), streamStarted) }, ) { continue } @@ -922,6 +966,11 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { }) } if err != nil { + if ciphertextFailoverErr != nil && service.OpenAICompactKeepaliveAdjustedWrittenSize(c) != writerSizeBeforeForward { + // This account read the ciphertext and was answering when the + // attempt failed: the conversation stays with it. + ciphertextAccepted = true + } if result != nil && result.ClientDisconnect { reqLog.Info("openai.client_disconnected", zap.Int64("account_id", account.ID), @@ -997,16 +1046,21 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { h.handleFailoverExhausted(c, failoverErr, streamStarted) return } - retryAction := retryState.HandleHTTP( - c.Request.Context(), - h.gatewayService, - account, - account.GetMappedModel(routingModel), - failoverErr, - true, - sameAccountRetryDelay, - "responses", - ) + // An account mismatch says nothing against the account: no + // same-account retry and no cooldown, the next account is tried. + retryAction := openAIFailoverRetrySwitchAccount + if !failoverErr.IsOpenAICiphertextAccountMismatch() { + retryAction = retryState.HandleHTTP( + c.Request.Context(), + h.gatewayService, + account, + account.GetMappedModel(routingModel), + failoverErr, + true, + sameAccountRetryDelay, + "responses", + ) + } h.finalizeOpenAIHTTPFailoverSelection(c, selection, account, account.GetMappedModel(routingModel), failoverErr, retryAction) switch retryAction { case openAIFailoverRetryReselect: @@ -1033,13 +1087,25 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { AllowCompatibleFallback: true, } } + if failoverErr.IsOpenAICiphertextAccountMismatch() { + if ciphertextFailoverErr == nil { + ciphertextFailoverErr = failoverErr + ciphertextFirstAccountID, ciphertextSessionHash = account.ID, sessionHash + } + if ciphertextSwitchCount >= maxOpenAICiphertextAccountSwitches { + h.handleFailoverExhausted(c, ciphertextFailoverErr, streamStarted) + return + } + ciphertextSwitchCount++ + service.ResetOpenAIReasoningRecoveryAttempt(c) + } if switchCount >= maxAccountSwitches { - h.handleFailoverExhausted(c, failoverErr, streamStarted) + h.handleFailoverExhausted(c, exhaustedFailoverErr(failoverErr), streamStarted) return } switchCount++ if h.gatewayService.ShouldStopOpenAIOAuth429Failover(account, failoverErr.StatusCode, switchCount, &oauth429FailoverState) { - h.handleFailoverExhausted(c, failoverErr, streamStarted) + h.handleFailoverExhausted(c, exhaustedFailoverErr(failoverErr), streamStarted) return } h.gatewayService.RecordOpenAIAccountSwitch() @@ -1048,6 +1114,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { zap.Int("upstream_status", failoverErr.StatusCode), zap.Int("switch_count", switchCount), zap.Int("max_switches", maxAccountSwitches), + zap.Bool("ciphertext_account_mismatch", failoverErr.IsOpenAICiphertextAccountMismatch()), } if account.Proxy != nil { failoverSwitchFields = append(failoverSwitchFields, @@ -1100,6 +1167,15 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { if shouldSubmitOpenAIUsage(err, result) { submitResponsesUsage(result) } + ciphertextAccepted = true + if ciphertextFailoverErr != nil { + // Selection usually bound the session to this account already. Where + // it leaves a binding in place, move one that still points at the + // account that could not read the conversation. + if bindErr := h.gatewayService.MoveStickySessionIfBoundTo(c.Request.Context(), apiKey.GroupID, sessionHash, account.ID, map[int64]struct{}{ciphertextFirstAccountID: {}}); bindErr != nil { + reqLog.Warn("openai.ciphertext_mismatch_session_bind_failed", zap.Int64("account_id", account.ID), zap.Error(bindErr)) + } + } reqLog.Debug("openai.request_completed", zap.Int64("account_id", account.ID), zap.Int("switch_count", switchCount), diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 68f39ca5d5c0..5c67ad4c3f63 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -746,6 +746,7 @@ type UpstreamFailoverError struct { ClientErrorParam string ClientMessage string SuppressAccountHealthPenalty bool + CiphertextAccountMismatch bool // 本账号无法校验请求里的密文且无法就地修复；原样请求可能被签发该密文的组织下的其他账号接受 } func (e *UpstreamFailoverError) Error() string { diff --git a/backend/internal/service/openai_continuation_diagnostic.go b/backend/internal/service/openai_continuation_diagnostic.go index 1584cfebcc61..ee07248b5b38 100644 --- a/backend/internal/service/openai_continuation_diagnostic.go +++ b/backend/internal/service/openai_continuation_diagnostic.go @@ -47,6 +47,10 @@ type openAIContinuationRecoveryShape struct { RetryAttempted bool `json:"retry_attempted"` Disposition string `json:"disposition"` NotAttemptedReason string `json:"not_attempted_reason,omitempty"` + // AccountMismatch marks an attempt in which the account could not read + // ciphertext of the request and could not repair it. Such a request is + // handed to another account when one is left to try. + AccountMismatch bool `json:"account_mismatch,omitempty"` } // Bytes, Characters and SHA256 always describe the complete original value. diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 9fdbd4d15943..89cc3e30e225 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1111,9 +1111,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco resp.Body = io.NopCloser(bytes.NewReader(respBody)) continuationStateError := classifyOpenAIContinuationStateError(upstreamMsg, respBody) if continuationStateError != openAIContinuationStateErrorNone { - // The failure describes request history, not account health. Do not - // let a compatibility proxy's 4xx/5xx wrapper fan this one request - // out across the remaining scheduler pool. + // The failure describes request history, not account health, whatever + // status a compatibility proxy wraps it in. Whether rejected + // ciphertext is offered to another account is decided once, when the + // attempt ends (StopError). s.recordOpenAIRequestTerminalUpstreamError(ctx, c, account, resp.StatusCode, resp.Header, body, respBody, "continuation_state", false, buildOpenAIContinuationDiagnostic(c, diagnosticIncomingBody, upstreamReq, body, respBody, string(continuationStateError)), diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 429d9a0e4954..1657819ad2ec 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -229,6 +229,26 @@ func (s *OpenAIGatewayService) BindStickySession(ctx context.Context, groupID *i return s.setStickySessionAccountID(ctx, groupID, sessionHash, accountID, ttl) } +// MoveStickySessionIfBoundTo binds the session to accountID only when it is +// bound to one of from at this moment. A binding that points anywhere else, +// and a session without a binding, are left as they are: the scheduler keeps a +// session on its account through a request it routes elsewhere, and only a +// binding the caller knows it moved may be moved again. +func (s *OpenAIGatewayService) MoveStickySessionIfBoundTo(ctx context.Context, groupID *int64, sessionHash string, accountID int64, from map[int64]struct{}) error { + if sessionHash == "" || accountID <= 0 || len(from) == 0 { + return nil + } + current, err := s.getStickySessionAccountID(ctx, groupID, sessionHash) + if err != nil || current <= 0 { + // Without a readable binding nothing is known to have been moved. + return nil + } + if _, moved := from[current]; !moved { + return nil + } + return s.BindStickySession(ctx, groupID, sessionHash, accountID) +} + // SelectAccount selects an OpenAI account with sticky session support func (s *OpenAIGatewayService) SelectAccount(ctx context.Context, groupID *int64, sessionHash string) (*Account, error) { return s.SelectAccountForModel(ctx, groupID, sessionHash, "") diff --git a/backend/internal/service/openai_gateway_service_hotpath_test.go b/backend/internal/service/openai_gateway_service_hotpath_test.go index 083ce4c1e1be..0c5800d40166 100644 --- a/backend/internal/service/openai_gateway_service_hotpath_test.go +++ b/backend/internal/service/openai_gateway_service_hotpath_test.go @@ -812,7 +812,7 @@ func TestOpenAIGatewayService_Forward_StreamingResponseFailedInvalidEncryptedCon require.NotContains(t, rec.Body.String(), "[DONE]") } -func TestOpenAIGatewayService_Forward_ContinuationStateWithToolOutputStopsWithoutReplay(t *testing.T) { +func TestOpenAIGatewayService_Forward_ContinuationStateWithToolOutputIsNotReplayedOnTheAccount(t *testing.T) { gin.SetMode(gin.TestMode) upstream := &httpUpstreamRecorder{ resp: &http.Response{ @@ -847,7 +847,7 @@ func TestOpenAIGatewayService_Forward_ContinuationStateWithToolOutputStopsWithou var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) require.True(t, failoverErr.IsOpenAIContinuationStateUnavailable()) - require.False(t, failoverErr.ShouldRetryNextAccount()) + require.True(t, failoverErr.IsOpenAICiphertextAccountMismatch(), "another account may accept the unchanged request") require.True(t, failoverErr.SuppressAccountHealthPenalty) require.Len(t, upstream.bodies, 1, "tool-output continuation must not be rewritten or replayed") require.Equal(t, "gAAA", gjson.GetBytes(upstream.bodies[0], "input.0.encrypted_content").String()) diff --git a/backend/internal/service/openai_gateway_upstream_errors.go b/backend/internal/service/openai_gateway_upstream_errors.go index 4d409f9d7d6f..6617b8d94b1f 100644 --- a/backend/internal/service/openai_gateway_upstream_errors.go +++ b/backend/internal/service/openai_gateway_upstream_errors.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "strings" + "time" "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" @@ -258,10 +259,9 @@ func (s *OpenAIGatewayService) shouldFailoverOpenAIUpstreamResponse(account *Acc if isOpenAIRequestScopedSafetyRejection(upstreamBody) { return false } - // A continuation-state rejection is request-scoped: choosing a different - // account cannot validate a previous response or provider-specific encrypted - // reasoning item. The caller returns a bounded recovery or a terminal client - // error before account-health side effects run. + // A continuation-state rejection is request-scoped and never an account + // failure. The caller returns a bounded recovery, an account mismatch or a + // terminal client error before account-health side effects run. if isOpenAIContinuationStateError(upstreamMsg, upstreamBody) { return false } @@ -378,8 +378,10 @@ const openAIRequestBodyTooLargeReason = GatewayFailureReason("openai_request_bod // OpenAIContinuationStateUnavailableCode identifies a request whose previous // response or encrypted reasoning state is no longer valid on its upstream. -// It is deliberately non-retryable: retrying on another account can only make -// the same account-bound state failure fan out across the scheduler pool. +// A lost previous response is bound to its account and ends the request. +// Rejected ciphertext is offered to a bounded number of other accounts first +// (openAICiphertextAccountMismatch); the client receives this code only when +// none accepts. const OpenAIContinuationStateUnavailableCode = "continuation_state_unavailable" // OpenAIContinuationStateUnavailableClientMessage is safe to surface to a @@ -642,12 +644,61 @@ func NewOpenAIContinuationStoreUnavailableError() *UpstreamFailoverError { return err } -// IsOpenAIContinuationStateUnavailable reports a continuation-state failure -// that must not switch accounts or reduce selected-account health. +// IsOpenAIContinuationStateUnavailable reports a continuation-state failure. +// It never reduces selected-account health, and it stops the request unless it +// is marked as an account mismatch. func (e *UpstreamFailoverError) IsOpenAIContinuationStateUnavailable() bool { return e != nil && e.Reason == openAIContinuationStateUnavailableReason } +// openAICiphertextAccountMismatch marks the request-scoped failure of one +// account as an account mismatch: the request carries ciphertext this account +// cannot verify and nothing on this account repairs it. Ciphertext is readable +// only by the upstream organisation that issued it, so the unchanged request +// may still be accepted by another account. The caller tries the next account; +// when none accepts, the failure renders with its own classification, exactly +// as it would have without the mark. +func openAICiphertextAccountMismatch(failure *UpstreamFailoverError) *UpstreamFailoverError { + marked := *failure + marked.ResponseHeaders = failure.ResponseHeaders.Clone() + marked.CiphertextAccountMismatch = true + marked.NextAccountAction = NextAccountRetry + marked.Scope = GatewayFailureScopeRequest + marked.SuppressAccountHealthPenalty = true + marked.RetryableOnSameAccount = false + marked.SameAccountRetryDelay = 0 + marked.SameAccountRetryDeadline = time.Time{} + marked.SameAccountRetryMax = 0 + return &marked +} + +// IsOpenAICiphertextAccountMismatch reports a failure that another account +// may not have for the same request. +func (e *UpstreamFailoverError) IsOpenAICiphertextAccountMismatch() bool { + return e != nil && e.CiphertextAccountMismatch +} + +// openAIUpstreamReportsUndecryptableCiphertext reads an upstream failure for a +// statement that ciphertext in the request could not be decrypted or verified. +// Besides the continuation-state classification it accepts the wording used +// for inter-agent message bodies and encrypted tool arguments, which arrives +// without a stable code. +func openAIUpstreamReportsUndecryptableCiphertext(payload []byte) bool { + message := extractOpenAISSEErrorMessage(payload) + switch classifyOpenAIContinuationStateError(message, payload) { + case openAIContinuationStateErrorInvalidEncryptedContent, openAIContinuationStateErrorThinkingSignatureInvalid: + return true + } + return strings.Contains(strings.ToLower(message), "could not be decrypted") +} + +// openAIRequestHoldsServerContext reports a request whose history lives with +// the upstream account that created it. +func openAIRequestHoldsServerContext(body []byte) bool { + conversation := gjson.GetBytes(body, "conversation") + return gjson.GetBytes(body, "previous_response_id").String() != "" || (conversation.Exists() && conversation.Type != gjson.Null) +} + func marshalOpenAIUpstreamJSON(v any) ([]byte, error) { var buf bytes.Buffer enc := json.NewEncoder(&buf) diff --git a/backend/internal/service/openai_oauth_passthrough_test.go b/backend/internal/service/openai_oauth_passthrough_test.go index c263f98823b7..daf10ef968e6 100644 --- a/backend/internal/service/openai_oauth_passthrough_test.go +++ b/backend/internal/service/openai_oauth_passthrough_test.go @@ -2126,13 +2126,17 @@ func TestOpenAIGatewayService_OpenAIPassthrough_InvalidEncryptedContentFirstErro require.Empty(t, repo.overloadCalls) } -func TestOpenAIGatewayService_OpenAIPassthrough_ContinuationErrorsStopBeforeAccountSideEffects(t *testing.T) { +func TestOpenAIGatewayService_OpenAIPassthrough_ContinuationErrorsHaveNoAccountSideEffects(t *testing.T) { gin.SetMode(gin.TestMode) tests := []struct { name string statusCode int responseBody string requestBody string + // accountMismatch: the rejection concerns ciphertext another account may + // read. A lost previous response, or a status that concerns the + // account's own access, ends the request on this account. + accountMismatch bool }{ { name: "structured previous response 400", @@ -2153,10 +2157,11 @@ func TestOpenAIGatewayService_OpenAIPassthrough_ContinuationErrorsStopBeforeAcco requestBody: `{"model":"gpt-5.2","stream":false,"previous_response_id":"resp_stale","input":"hi"}`, }, { - name: "narrow invalid encrypted message 503", - statusCode: http.StatusServiceUnavailable, - responseBody: `{"error":{"message":"Encrypted content could not be verified"}}`, - requestBody: `{"model":"gpt-5.2","stream":false,"input":"hi"}`, + name: "narrow invalid encrypted message 503", + statusCode: http.StatusServiceUnavailable, + responseBody: `{"error":{"message":"Encrypted content could not be verified"}}`, + requestBody: `{"model":"gpt-5.2","stream":false,"input":"hi"}`, + accountMismatch: true, }, } @@ -2200,7 +2205,8 @@ func TestOpenAIGatewayService_OpenAIPassthrough_ContinuationErrorsStopBeforeAcco require.True(t, failoverErr.IsOpenAIContinuationStateUnavailable()) require.Equal(t, http.StatusBadRequest, failoverErr.ClientStatusCode) require.Equal(t, GatewayFailureScopeRequest, failoverErr.Scope) - require.False(t, failoverErr.ShouldRetryNextAccount()) + require.Equal(t, tt.accountMismatch, failoverErr.IsOpenAICiphertextAccountMismatch()) + require.Equal(t, tt.accountMismatch, failoverErr.ShouldRetryNextAccount()) require.True(t, failoverErr.SuppressAccountHealthPenalty) require.Len(t, upstream.bodies, 1, "unsafe continuation requests must not be replayed") require.False(t, c.Writer.Written(), "continuation classification must happen before downstream response commitment") @@ -2212,6 +2218,13 @@ func TestOpenAIGatewayService_OpenAIPassthrough_ContinuationErrorsStopBeforeAcco events, ok := value.([]*OpsUpstreamErrorEvent) require.True(t, ok) require.NotEmpty(t, events) + if tt.accountMismatch { + // The hand-off to another account is one more event of the attempt. + handOff := events[len(events)-1] + require.Equal(t, "reasoning_recovery", handOff.Kind) + require.True(t, handOff.ContinuationDiagnostic.Recovery.AccountMismatch) + events = events[:len(events)-1] + } require.Equal(t, "continuation_state", events[len(events)-1].Kind) require.True(t, events[len(events)-1].Passthrough) require.Equal(t, openAIOpsUpstreamErrorMessage([]byte(tt.responseBody)), events[len(events)-1].Message) diff --git a/backend/internal/service/openai_reasoning_recovery.go b/backend/internal/service/openai_reasoning_recovery.go index 0b74faad6dc1..33733ceb376e 100644 --- a/backend/internal/service/openai_reasoning_recovery.go +++ b/backend/internal/service/openai_reasoning_recovery.go @@ -39,7 +39,10 @@ const ( ) // openAIReasoningRecoveryState owns the single, explicitly lossy retry for an -// HTTP Responses request. It never re-enters a protocol converter or scheduler. +// HTTP Responses request on one account. It never re-enters a protocol +// converter or scheduler itself; when the account cannot serve the request +// because of ciphertext it cannot verify, StopError says so and the handler +// decides whether another account is tried. // original is immutable; wire is the exact payload of the latest sent attempt. type openAIReasoningRecoveryState struct { ctx context.Context @@ -75,6 +78,12 @@ type openAIReasoningRecoveryState struct { // redactPayload applies the upstream Agent Identity credential redaction // before an upstream payload is recorded for administrators. redactPayload func([]byte) []byte + + // switchAccounts is the global switch for this request, whether or not + // this account type supports the stripped retry. + switchAccounts bool + // accountMismatch records that this attempt ended as an account mismatch. + accountMismatch bool } func (s *OpenAIGatewayService) newOpenAIReasoningRecoveryState(ctx context.Context, c *gin.Context, account *Account, token string) *openAIReasoningRecoveryState { @@ -82,10 +91,11 @@ func (s *OpenAIGatewayService) newOpenAIReasoningRecoveryState(ctx context.Conte ctx = context.Background() } // The global switch is read here, once for this forwarding attempt. + switchedOn := s.reasoningRecovery.Enabled() && !isOpenAICompatMessagesBridgeContext(c) r := &openAIReasoningRecoveryState{ ctx: ctx, c: c, account: account, token: token, - enabled: s.reasoningRecovery.Enabled() && account.supportsOpenAIReasoningRecovery() && !isOpenAICompatMessagesBridgeContext(c), - store: s.openAIReasoningStateStore(), budget: openAIReasoningCacheBudgetForRequest(c), + enabled: switchedOn && account.supportsOpenAIReasoningRecovery(), switchAccounts: switchedOn, + store: s.openAIReasoningStateStore(), budget: openAIReasoningCacheBudgetForRequest(c), } r.redactPayload = func(payload []byte) []byte { return s.redactAgentIdentitySensitiveBody(ctx, account, payload) } if c != nil { @@ -601,6 +611,8 @@ func openAIReasoningRecoveryProtectedStatus(status int) bool { // and account-health accounting, but deliberately does not implement Unwrap. // A caller must handle this terminal explicitly, never feed Failure back into // scheduling: the single same-source recovery budget has already been spent. +// An attempt that ended as an account mismatch is not a terminal; StopError +// returns its failure marked by openAICiphertextAccountMismatch instead. type OpenAIReasoningRecoveryTerminalError struct { Failure *UpstreamFailoverError // FailureTerminalForwarded is explicit parser/write evidence, not a claim @@ -677,7 +689,8 @@ func (r *openAIReasoningRecoveryState) StopError(err error) error { failure = r.requestRejectionFromStream(r.failurePayload) } cacheSkipRejected := r.cacheSkippedItems > 0 && failure != nil && failure.IsOpenAIRequestRejected() - if !r.retryUsed && !signatureRejected && !cacheSkipRejected { + r.accountMismatch = r.anotherAccountMayAccept(failure, signatureRejected) + if !r.retryUsed && !signatureRejected && !cacheSkipRejected && !r.accountMismatch { return err } if failure == nil { @@ -697,6 +710,9 @@ func (r *openAIReasoningRecoveryState) StopError(err error) error { } r.record(action, r.upstreamStatus(failure.StatusCode), rejection.code, r.failurePayload, 0) } + if r.accountMismatch { + return openAICiphertextAccountMismatch(failure) + } if !r.retryUsed && !r.semanticCommitted { return failure } @@ -705,6 +721,44 @@ func (r *openAIReasoningRecoveryState) StopError(err error) error { return terminal } +// anotherAccountMayAccept decides whether this attempt ended as an account +// mismatch: the upstream said it could not decrypt or verify ciphertext in the +// request, on the first send or on the stripped one, nothing on this account +// repaired the request, and no output or failure reached the client. Any other +// rejection of the request would be the same on every account. The handler +// still confirms the client-side state before it switches. +func (r *openAIReasoningRecoveryState) anotherAccountMayAccept(failure *UpstreamFailoverError, signatureRejected bool) bool { + if !r.switchAccounts || r.semanticCommitted || r.failureTerminalForwarded || r.ctx.Err() != nil || + openAIReasoningRecoveryProtectedStatus(r.responseStatus) || + // wire is absent when this account type has no stripped retry. + openAIRequestHoldsServerContext(r.wire) || openAIRequestHoldsServerContext(r.diagnosticIncoming) { + return false + } + if r.retryUsed && !r.retryDispatched { + // The stripped retry was prepared but never sent. That is a fault in + // how this request was built, not evidence about the account. + return false + } + switch { + case failure == nil: + return signatureRejected + case !failure.IsOpenAIContinuationStateUnavailable() && !failure.IsOpenAIRequestRejected(): + return false + default: + return signatureRejected || openAIUpstreamReportsUndecryptableCiphertext(r.failurePayload) + } +} + +// ResetOpenAIReasoningRecoveryAttempt detaches the state of a rejected attempt +// before the request moves to another account, so that a forwarder which +// creates no state of its own does not read this one. The response header that +// names the last recovery action is left as it is. +func ResetOpenAIReasoningRecoveryAttempt(c *gin.Context) { + if c != nil { + c.Set(openAIReasoningRecoveryContextKey, (*openAIReasoningRecoveryState)(nil)) + } +} + func (r *openAIReasoningRecoveryState) upstreamStatus(fallback int) int { if r != nil && r.responseStatus > 0 { return r.responseStatus @@ -729,6 +783,7 @@ func (r *openAIReasoningRecoveryState) continuationDiagnosticRecovery(payload [] CacheSkippedItems: r.cacheSkippedItems, RetryAttempted: r.retryDispatched, Disposition: r.diagnosticState, + AccountMismatch: r.accountMismatch, } if shape.Disposition == "" { shape.Disposition = "not_attempted" diff --git a/backend/internal/service/openai_reasoning_recovery_terminal_test.go b/backend/internal/service/openai_reasoning_recovery_terminal_test.go index b8ffa76db22b..14a7e9460d1b 100644 --- a/backend/internal/service/openai_reasoning_recovery_terminal_test.go +++ b/backend/internal/service/openai_reasoning_recovery_terminal_test.go @@ -57,7 +57,11 @@ func lastReasoningRecoveryDiagnostic(t *testing.T, c *gin.Context) *OpsUpstreamE return last } -func TestOpenAIReasoningRecoveryTerminalsPreserveClassificationAndDiagnostic(t *testing.T) { +// Every outcome keeps its classification and diagnostic. Where the upstream +// said it could not verify ciphertext, the attempt ends as an account mismatch, +// a failure the handler may hand to another account; a validation rejection is +// the same on every account and ends where it happened. +func TestOpenAIReasoningRecoveryOutcomesPreserveClassificationAndDiagnostic(t *testing.T) { const signature = `{"code":"thinking_signature_invalid","param":"input[1].encrypted_content","message":"private-signature-detail-8162"}` const wrongTarget = `{"code":"thinking_signature_invalid","param":"input[2].encrypted_content","message":"private-signature-detail-8162"}` const validation = `{"type":"invalid_request_error","code":"missing_required_parameter","param":"input[3].call_id","message":"private-validation-detail-7514"}` @@ -109,22 +113,28 @@ func TestOpenAIReasoningRecoveryTerminalsPreserveClassificationAndDiagnostic(t * result, err := svc.Forward(context.Background(), c, account, []byte(input)) require.Error(t, err) require.Nil(t, result) + mismatch := classification != "request_validation" var failure *UpstreamFailoverError - if retried { - var stopped *OpenAIReasoningRecoveryTerminalError + var stopped *OpenAIReasoningRecoveryTerminalError + if retried && !mismatch { require.ErrorAs(t, err, &stopped) require.False(t, errors.As(err, &failure), "spent recovery must not reenter failover") failure = stopped.Failure - require.Len(t, upstream.requests, 2, "no third POST") } else { + require.False(t, errors.As(err, &stopped), "an account mismatch is not a terminal") require.ErrorAs(t, err, &failure) + } + if retried { + require.Len(t, upstream.requests, 2, "no third POST on this account") + } else { require.Len(t, upstream.requests, 1) } require.Equal(t, status, failure.StatusCode, "retain actual HTTP status separately from client semantics") require.Equal(t, http.StatusBadRequest, failure.ClientStatusCode) require.Equal(t, classification == "request_validation", failure.IsOpenAIRequestRejected()) require.Equal(t, classification != "request_validation", failure.IsOpenAIContinuationStateUnavailable()) - require.False(t, failure.ShouldRetryNextAccount()) + require.Equal(t, mismatch, failure.IsOpenAICiphertextAccountMismatch()) + require.Equal(t, mismatch, failure.ShouldRetryNextAccount()) require.False(t, failure.RetryableOnSameAccount) require.True(t, failure.SuppressAccountHealthPenalty) require.False(t, failure.ShouldReportAccountScheduleFailure()) @@ -146,6 +156,7 @@ func TestOpenAIReasoningRecoveryTerminalsPreserveClassificationAndDiagnostic(t * require.Equal(t, len(upstream.lastBody), diagnostic.Wire.BodyBytes) require.Equal(t, 1, diagnostic.Incoming.History.Encrypted) require.NotNil(t, diagnostic.Recovery) + require.Equal(t, mismatch, diagnostic.Recovery.AccountMismatch) require.Equal(t, retried, diagnostic.Recovery.RetryAttempted) if retried { require.Equal(t, "budget_exhausted", diagnostic.Recovery.Disposition) diff --git a/backend/internal/service/openai_reasoning_recovery_test.go b/backend/internal/service/openai_reasoning_recovery_test.go index a4c74624e95d..aff197419510 100644 --- a/backend/internal/service/openai_reasoning_recovery_test.go +++ b/backend/internal/service/openai_reasoning_recovery_test.go @@ -105,6 +105,79 @@ func TestOpenAIReasoningRecoveryNamedReasoningLeavesOtherCarriers(t *testing.T) } } +func TestOpenAIReasoningRecoveryStopErrorMarksAccountMismatch(t *testing.T) { + const signature = `{"error":{"message":"The encrypted content for item amsg_b could not be verified.","param":null,"code":"invalid_encrypted_content"}}` + const undecryptable = `{"error":{"type":"invalid_request_error","message":"Encrypted function output content could not be decrypted or decoded."}}` + const validation = `{"error":{"type":"invalid_request_error","code":"missing_required_parameter","message":"input[3].call_id is required"}}` + serverHeld := strings.Replace(reasoningRecoveryAgentMessageFixture, `{"model"`, `{"previous_response_id":"resp_old","model"`, 1) + tests := []struct { + name, body, payload string + status int + arrange func(*openAIReasoningRecoveryState) + want bool + }{ + {"ciphertext_the_strip_cannot_reach", reasoningRecoveryAgentMessageFixture, signature, http.StatusBadRequest, nil, true}, + {"undecryptable_without_a_code", reasoningRecoveryAgentMessageFixture, undecryptable, http.StatusBadRequest, nil, true}, + {"ordinary_validation", reasoningRecoveryAgentMessageFixture, validation, http.StatusBadRequest, nil, false}, + {"history_held_by_the_account", serverHeld, signature, http.StatusBadRequest, nil, false}, + {"history_held_by_an_account_without_the_stripped_retry", serverHeld, signature, http.StatusBadRequest, func(r *openAIReasoningRecoveryState) { r.enabled, r.wire = false, nil }, false}, + {"status_about_the_account", reasoningRecoveryAgentMessageFixture, signature, http.StatusForbidden, nil, false}, + {"switch_off", reasoningRecoveryAgentMessageFixture, signature, http.StatusBadRequest, func(r *openAIReasoningRecoveryState) { r.enabled, r.switchAccounts = false, false }, false}, + {"output_already_sent", reasoningRecoveryAgentMessageFixture, signature, http.StatusBadRequest, func(r *openAIReasoningRecoveryState) { r.semanticCommitted = true }, false}, + {"failure_already_forwarded", reasoningRecoveryAgentMessageFixture, signature, http.StatusBadRequest, func(r *openAIReasoningRecoveryState) { r.failureTerminalForwarded = true }, false}, + {"request_cancelled", reasoningRecoveryAgentMessageFixture, signature, http.StatusBadRequest, func(r *openAIReasoningRecoveryState) { + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + r.ctx = cancelled + }, false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + state, req, _ := newReasoningRecoveryTestState(t, context.Background(), test.body) + state.BindDiagnosticRequest([]byte(test.body), req) + _, retry := state.TryRecover(test.status, nil, []byte(test.payload), false) + require.False(t, retry) + if test.arrange != nil { + test.arrange(state) + } + var attempt error = NewOpenAIRequestRejectedError(test.status, nil) + if _, rejected := parseOpenAIReasoningRejection([]byte(test.payload)); rejected { + attempt = NewOpenAIContinuationStateUnavailableError(test.status, nil, []byte(test.payload)) + } + var failure *UpstreamFailoverError + stopped := state.StopError(attempt) + if !errors.As(stopped, &failure) { + var terminal *OpenAIReasoningRecoveryTerminalError + require.ErrorAs(t, stopped, &terminal) + failure = terminal.Failure + } + require.Equal(t, test.want, failure.IsOpenAICiphertextAccountMismatch()) + require.Equal(t, test.want, failure.ShouldRetryNextAccount()) + require.True(t, failure.SuppressAccountHealthPenalty) + if test.want { + // The hand-off is on the Ops record whichever way it was recognised. + events, _ := state.c.Get(OpsUpstreamErrorsKey) + recorded := events.([]*OpsUpstreamErrorEvent) + require.True(t, recorded[len(recorded)-1].ContinuationDiagnostic.Recovery.AccountMismatch) + } + var original *UpstreamFailoverError + require.ErrorAs(t, attempt, &original) + require.Equal(t, original.Reason, failure.Reason, "the mark keeps the failure's own classification") + require.False(t, original.IsOpenAICiphertextAccountMismatch(), "the attempt's own error value is not modified") + }) + } +} + +func TestResetOpenAIReasoningRecoveryAttemptDetachesTheState(t *testing.T) { + state, _, rec := newReasoningRecoveryTestState(t, context.Background(), reasoningRecoveryFixture) + _, retry := state.TryRecover(http.StatusBadRequest, nil, []byte(`{"error":{"code":"invalid_encrypted_content"}}`), false) + require.True(t, retry) + require.Same(t, state, openAIReasoningRecoveryStateFromContext(state.c)) + ResetOpenAIReasoningRecoveryAttempt(state.c) + require.Nil(t, openAIReasoningRecoveryStateFromContext(state.c), "the next account must not read this attempt") + require.Equal(t, "retry_without_encrypted_content", rec.Header().Get(openAIReasoningRecoveryHeader), "the header names the last recovery action of the request") +} + func TestOpenAIReasoningRecoveryPreservesEveryNonCipherField(t *testing.T) { state, _, rec := newReasoningRecoveryTestState(t, context.Background(), reasoningRecoveryFixture) before := bytes.Clone(state.wire) diff --git a/backend/internal/service/openai_sticky_compat_test.go b/backend/internal/service/openai_sticky_compat_test.go index d32115c38cf8..42007ae35387 100644 --- a/backend/internal/service/openai_sticky_compat_test.go +++ b/backend/internal/service/openai_sticky_compat_test.go @@ -115,3 +115,22 @@ func TestSnapshotOpenAICompatibilityFallbackMetrics(t *testing.T) { require.GreaterOrEqual(t, after.MetadataLegacyFallbackTotal, before.MetadataLegacyFallbackTotal+1) require.GreaterOrEqual(t, after.MetadataLegacyFallbackThinkingEnabledTotal, before.MetadataLegacyFallbackThinkingEnabledTotal+1) } + +func TestMoveStickySessionIfBoundTo_OnlyMovesABindingTheCallerMoved(t *testing.T) { + cache := &stubGatewayCache{sessionBindings: map[string]int64{"openai:conversation": 6}} + svc := &OpenAIGatewayService{cache: cache, cfg: &config.Config{}} + later := map[int64]struct{}{2: {}, 3: {}} + + // The scheduler kept the session on its own account while it routed this + // request elsewhere: that binding stays. + require.NoError(t, svc.MoveStickySessionIfBoundTo(context.Background(), nil, "conversation", 1, later)) + require.Equal(t, int64(6), cache.sessionBindings["openai:conversation"]) + + cache.sessionBindings["openai:conversation"] = 3 + require.NoError(t, svc.MoveStickySessionIfBoundTo(context.Background(), nil, "conversation", 1, later)) + require.Equal(t, int64(1), cache.sessionBindings["openai:conversation"]) + + delete(cache.sessionBindings, "openai:conversation") + require.NoError(t, svc.MoveStickySessionIfBoundTo(context.Background(), nil, "conversation", 1, later)) + require.NotContains(t, cache.sessionBindings, "openai:conversation", "a session without a binding gets none") +} diff --git a/backend/internal/service/openai_ws_protocol_forward_test.go b/backend/internal/service/openai_ws_protocol_forward_test.go index 462c42565655..0f325bacec7b 100644 --- a/backend/internal/service/openai_ws_protocol_forward_test.go +++ b/backend/internal/service/openai_ws_protocol_forward_test.go @@ -273,7 +273,7 @@ func TestOpenAIGatewayService_Forward_HTTPIngressInvalidEncryptedContentIsTermin require.Equal(t, "client_protocol_http", reason) } -func TestOpenAIGatewayService_Forward_HTTPIngressWrappedInvalidEncryptedContentIsTerminal(t *testing.T) { +func TestOpenAIGatewayService_Forward_HTTPIngressWrappedInvalidEncryptedContentIsNotRepairedOnTheAccount(t *testing.T) { gin.SetMode(gin.TestMode) wsFallbackServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) @@ -343,9 +343,9 @@ func TestOpenAIGatewayService_Forward_HTTPIngressWrappedInvalidEncryptedContentI var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) require.True(t, failoverErr.IsOpenAIContinuationStateUnavailable()) - require.False(t, failoverErr.ShouldRetryNextAccount()) + require.True(t, failoverErr.IsOpenAICiphertextAccountMismatch(), "原样请求可以交给其他账号") require.True(t, failoverErr.SuppressAccountHealthPenalty) - require.Equal(t, 1, upstream.callCount, "wrapped invalid_encrypted_content 也必须终止，不能删状态重试") + require.Equal(t, 1, upstream.callCount, "wrapped invalid_encrypted_content 在本账号上不能删状态重试") require.Len(t, upstream.bodies, 1) require.False(t, c.Writer.Written(), "continuation 应交由 handler 输出标准终态，不能伪造成功")