feat(reasoning): hand a request to another account when this one cannot read its ciphertext

.downstream/gpt6-models-and-prompts.md

@@ -66,7 +66,7 @@ Claude OAuth 伪装系统块只由上游设置决定：设置 → 网关的 `ena
 
 侧栏“扩展功能 → 推理恢复”（`/admin/reasoning-recovery`）：一个全局开关和保存按钮，默认开启，没有账号级选项；页面打开时读取一次，读取失败时显示错误、不显示开关。存为 `settings.reasoning_recovery_config` 一行 JSON（`{"enabled": <bool>}`），没有该行即开启；`GET/PUT /api/v1/admin/reasoning-recovery` 以标准响应封装读写 `{"enabled": <bool>}`，管理员认证、无二次验证，保存记入操作审计，请求体不是严格的 `{"enabled": <bool>}` 时返回 400 且不保存。请求路径只读内存中的布尔值（保存即替换，60 秒后台刷新以同步多实例），不查询数据库，每次转发开始时读取一次。启动时读取失败或存储值不是严格的 `{"enabled": <bool>}` 则停止启动；运行中刷新失败保留当前值。
 
-开启时，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 的请求只由客户端消息转换而来，网关不保存响应中的推理项。
 

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))
+}