fix(reasoning): recover when the upstream names a reasoning item; remember long histories

.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`，在同一账号重发一次；被拒密文按来源在 Redis 记忆 24 小时，期间同源请求发送前即去掉。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 预算。Messages 桥接与 WebSocket 不恢复。关闭时不重发也不预先去除，签名拒绝仍作为请求级终态返回，不切换账号。
 
 迁移 `267_purge_account_reasoning_options.sql` 一次性删除各账号 `extra` 中原有的两个账号级推理选项键（键名见该文件）；此后没有代码读写它们，已存的账号级取值不带入全局开关。Chat Completions 转 Responses 的请求只由客户端消息转换而来，网关不保存响应中的推理项。
 

backend/internal/codexruntime/recovery/policy.go

@@ -50,7 +50,9 @@ func Select(in extensionv1.RecoverySelectionQuery) extensionv1.RecoverySelection
 	if in.ServerContext {
 		return extensionv1.RecoverySelection{Reason: "server_held_context"}
 	}
-	if in.EncryptedFields != len(in.CipherIndices) {
+	// Another ciphertext carrier (compaction, an inter-agent message) may be the
+	// rejected one, unless the upstream names a reasoning item itself.
+	if in.EncryptedFields != len(in.CipherIndices) && !in.NamedCipher {
 		return extensionv1.RecoverySelection{Reason: "ambiguous_encrypted_carriers"}
 	}
 	return extensionv1.RecoverySelection{Indices: append([]int(nil), in.CipherIndices...), Reason: "recovery_not_dispatched"}

backend/internal/codexruntime/recovery/policy_test.go

@@ -38,4 +38,8 @@ func TestRecoveryPolicyRequiresExactErrorAndUnambiguousLocalCipherScope(t *testi
 	if Select(query).Reason != "ambiguous_encrypted_carriers" {
 		t.Fatal("compaction ciphertext may be affected")
 	}
+	query.NamedCipher = true
+	if len(Select(query).Indices) != 2 {
+		t.Fatal("a rejection that names a reasoning item is not ambiguous")
+	}
 }

backend/internal/nativeapi/codex_recovery.go

@@ -28,6 +28,9 @@ type RecoverySelectionQuery struct {
 	EncryptedFields  int    `json:"encrypted_fields"`
 	ServerContext    bool   `json:"server_context"`
 	Param            string `json:"param"`
+	// NamedCipher reports that the upstream error names, by item id, an item
+	// listed in CipherIndices.
+	NamedCipher bool `json:"named_cipher"`
 }
 
 type RecoverySelection struct {

backend/internal/service/openai_continuation_diagnostic.go

@@ -103,6 +103,9 @@ type openAIContinuationHistory struct {
 	CallIDsSHA256      string `json:"call_ids_sha256,omitempty"`
 	OutputIDsSHA256    string `json:"output_ids_sha256,omitempty"`
 	EncryptedSHA256    string `json:"encrypted_sha256,omitempty"`
+	// NestedEncrypted counts ciphertext held in an item's content parts, such
+	// as an inter-agent message body; Encrypted counts the item-level field.
+	NestedEncrypted int `json:"nested_encrypted"`
 }
 
 func buildOpenAIContinuationDiagnostic(c *gin.Context, incomingBody []byte, upstreamReq *http.Request, preparedBody, upstreamError []byte, classification string) *OpenAIContinuationDiagnostic {
@@ -290,6 +293,14 @@ func continuationDiagnosticHistory(input gjson.Result) openAIContinuationHistory
 			shape.Encrypted++
 			continuationDiagnosticHashPart(encryptedDigest, encrypted.String())
 		}
+		if content := item.Get("content"); content.IsArray() {
+			content.ForEach(func(_, part gjson.Result) bool {
+				if nested := part.Get("encrypted_content"); nested.Type == gjson.String && nested.String() != "" {
+					shape.NestedEncrypted++
+				}
+				return true
+			})
+		}
 		return true
 	})
 	for id, count := range calls {

backend/internal/service/openai_continuation_diagnostic_test.go

@@ -356,6 +356,13 @@ func (r *continuationDiagnosticTestReadCloser) Close() error {
 	return nil
 }
 
+func TestOpenAIContinuationDiagnosticCountsNestedCiphertext(t *testing.T) {
+	history := continuationDiagnosticHistory(gjson.Parse(`[{"type":"reasoning","encrypted_content":"a"},{"type":"agent_message","content":[{"type":"input_text","text":"Payload:"},{"type":"encrypted_content","encrypted_content":"b"}]},{"type":"message","role":"user","content":"text"}]`))
+	if history.Other != 1 || history.Encrypted != 1 || history.NestedEncrypted != 1 {
+		t.Fatalf("other=%d encrypted=%d nested=%d", history.Other, history.Encrypted, history.NestedEncrypted)
+	}
+}
+
 func TestOpenAIContinuationDiagnosticResourceBounds(t *testing.T) {
 	t.Run("history_scan_has_a_hard_limit", func(t *testing.T) {
 		body := []byte(`{"input":[` + strings.TrimSuffix(strings.Repeat(`{"type":"function_call","call_id":"private-large-history-id-57319"},`, 1100), ",") + `]}`)

backend/internal/service/openai_reasoning_recovery.go

@@ -9,7 +9,9 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"maps"
 	"net/http"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -329,11 +331,18 @@ func (r *openAIReasoningRecoveryState) skipRejectedHistory(body []byte) []byte {
 	for _, item := range items {
 		hashes = append(hashes, item.hash)
 	}
-	var rejected map[string]OpenAIRejectedReasoning
+	// The store takes a bounded number of hashes per call; a long history is
+	// looked up in batches inside the one I/O budget.
+	rejected := make(map[string]OpenAIRejectedReasoning, len(hashes))
 	err := r.budget.Do(r.ctx, func(ioCtx context.Context) error {
-		var readErr error
-		rejected, readErr = r.store.GetOpenAIRejectedReasoning(ioCtx, r.scope, hashes)
-		return readErr
+		for batch := range slices.Chunk(hashes, OpenAIReasoningStateMaxLookupEntries) {
+			found, readErr := r.store.GetOpenAIRejectedReasoning(ioCtx, r.scope, batch)
+			if readErr != nil {
+				return readErr
+			}
+			maps.Copy(rejected, found)
+		}
+		return nil
 	})
 	if err != nil {
 		return body
@@ -361,6 +370,7 @@ func (r *openAIReasoningRecoveryState) skipRejectedHistory(body []byte) []byte {
 type openAIReasoningCipherItem struct {
 	index int
 	hash  string
+	id    string
 }
 
 func openAIReasoningCipherItems(body []byte) []openAIReasoningCipherItem {
@@ -372,17 +382,26 @@ func openAIReasoningCipherItems(body []byte) []openAIReasoningCipherItem {
 	for index, item := range input.Array() {
 		cipher := item.Get("encrypted_content")
 		if item.Get("type").String() == "reasoning" && cipher.Type == gjson.String && cipher.String() != "" {
-			out = append(out, openAIReasoningCipherItem{index, openAIReasoningDigest([]byte(cipher.String()))})
+			entry := openAIReasoningCipherItem{index: index, hash: openAIReasoningDigest([]byte(cipher.String()))}
+			if id := item.Get("id"); id.Type == gjson.String {
+				entry.id = id.String()
+			}
+			out = append(out, entry)
 		}
 	}
 	return out
 }
 
-type openAIReasoningRejection struct{ code, param string }
+// message is the upstream error text; selection reads it only for the id of
+// the rejected item.
+type openAIReasoningRejection struct{ code, param, message string }
 
 func parseOpenAIReasoningRejection(payload []byte) (openAIReasoningRejection, bool) {
 	result := codexrecovery.Rejection(openAIRecoveryEnvelope(payload))
-	return openAIReasoningRejection{code: result.Code, param: result.Param}, result.Recognized
+	if !result.Recognized {
+		return openAIReasoningRejection{}, false
+	}
+	return openAIReasoningRejection{code: result.Code, param: result.Param, message: openAIRecoveryErrorMessage(payload)}, true
 }
 
 func openAIReasoningToolHistoryAllowsRecovery(body []byte) bool {
@@ -396,7 +415,7 @@ func openAIReasoningToolHistoryAllowsRecovery(body []byte) bool {
 }
 
 func openAIReasoningRejectedIndices(body []byte, rejection openAIReasoningRejection) ([]int, []string) {
-	selection := selectOpenAIRecoveryIndices(body, rejection.param)
+	selection := selectOpenAIRecoveryIndices(body, rejection)
 	if len(selection.Indices) == 0 {
 		return nil, nil
 	}
@@ -557,7 +576,12 @@ func (r *openAIReasoningRecoveryState) TryRecover(status int, headers http.Heade
 	r.diagnosticState = "retry_prepared"
 	if r.store != nil && r.scope.ScopeHash != "" {
 		_ = r.budget.Do(r.ctx, func(ioCtx context.Context) error {
-			return r.store.PutOpenAIRejectedReasoning(ioCtx, r.scope, hashes)
+			for batch := range slices.Chunk(hashes, OpenAIReasoningStateMaxLookupEntries) {
+				if err := r.store.PutOpenAIRejectedReasoning(ioCtx, r.scope, batch); err != nil {
+					return err
+				}
+			}
+			return nil
 		})
 	}
 	r.record("retry_without_encrypted_content", status, rejection.code, payload, len(indices))
@@ -735,7 +759,7 @@ func (r *openAIReasoningRecoveryState) recoveryNotAttemptedReason(payload []byte
 	if !ok {
 		return "not_signature_rejection"
 	}
-	selection := selectOpenAIRecoveryIndices(r.wire, rejection.param)
+	selection := selectOpenAIRecoveryIndices(r.wire, rejection)
 	if len(selection.Indices) > 0 {
 		if _, err := stripOpenAIReasoningCipherIndices(r.wire, selection.Indices); err != nil {
 			return "rewrite_failed"

backend/internal/service/openai_reasoning_recovery_policy.go

@@ -2,12 +2,25 @@ package service
 
 import (
 	"encoding/json"
+	"strings"
 
 	codexrecovery "github.com/Wei-Shaw/sub2api/internal/codexruntime/recovery"
 	extensionv1 "github.com/Wei-Shaw/sub2api/internal/nativeapi"
 	"github.com/tidwall/gjson"
 )
 
+// openAIRecoveryErrorSource locates the error object of an upstream payload.
+func openAIRecoveryErrorSource(root gjson.Result) (string, gjson.Result) {
+	switch {
+	case root.Get("response.error").IsObject():
+		return "response", root.Get("response.error")
+	case root.Get("error").IsObject():
+		return "error", root.Get("error")
+	default:
+		return "root", root
+	}
+}
+
 // openAIRecoveryEnvelope reduces an upstream error payload to the protocol
 // shape the recovery policy decides on.
 func openAIRecoveryEnvelope(payload []byte) extensionv1.RecoveryEnvelope {
@@ -24,14 +37,7 @@ func openAIRecoveryEnvelope(payload []byte) extensionv1.RecoveryEnvelope {
 	}
 	envelope.Event, envelope.Status, envelope.ResponseStatus = root.Get("type").String(), root.Get("status").String(), root.Get("response.status").String()
 	var source gjson.Result
-	switch {
-	case root.Get("response.error").IsObject():
-		envelope.ErrorLocation, source = "response", root.Get("response.error")
-	case root.Get("error").IsObject():
-		envelope.ErrorLocation, source = "error", root.Get("error")
-	default:
-		envelope.ErrorLocation, source = "root", root
-	}
+	envelope.ErrorLocation, source = openAIRecoveryErrorSource(root)
 	if code := source.Get("code"); code.Type == gjson.String {
 		value := code.String()
 		envelope.Code = &value
@@ -42,15 +48,53 @@ func openAIRecoveryEnvelope(payload []byte) extensionv1.RecoveryEnvelope {
 	return envelope
 }
 
+// openAIRecoveryErrorMessage returns the message of the error object the
+// envelope takes its code and param from.
+func openAIRecoveryErrorMessage(payload []byte) string {
+	_, source := openAIRecoveryErrorSource(gjson.ParseBytes(payload))
+	if message := source.Get("message"); message.Type == gjson.String {
+		return message.String()
+	}
+	return ""
+}
+
+// openAIRecoveryMessageNamesItem reports whether the upstream message contains
+// a Responses reasoning item id as a whole identifier. The upstream reports a
+// ciphertext it cannot verify as "… for item rs_… could not be verified" with a
+// null param; the id is the only place the rejected item is identified.
+func openAIRecoveryMessageNamesItem(message, id string) bool {
+	if !strings.HasPrefix(id, "rs_") || len(id) == len("rs_") {
+		return false
+	}
+	identifier := func(b byte) bool {
+		return b == '_' || b == '-' || (b >= '0' && b <= '9') || (b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z')
+	}
+	for offset := 0; ; {
+		at := strings.Index(message[offset:], id)
+		if at < 0 {
+			return false
+		}
+		start := offset + at
+		end := start + len(id)
+		if (start == 0 || !identifier(message[start-1])) && (end == len(message) || !identifier(message[end])) {
+			return true
+		}
+		offset = start + 1
+	}
+}
+
 // selectOpenAIRecoveryIndices asks the recovery policy which reasoning items of
-// the sent body may lose their ciphertext for the rejection's param.
-func selectOpenAIRecoveryIndices(body []byte, param string) extensionv1.RecoverySelection {
-	query := extensionv1.RecoverySelectionQuery{Param: param}
+// the sent body may lose their ciphertext for the rejection.
+func selectOpenAIRecoveryIndices(body []byte, rejection openAIReasoningRejection) extensionv1.RecoverySelection {
+	query := extensionv1.RecoverySelectionQuery{Param: rejection.param}
 	if _, err := canonicalReasoningCacheJSON(body); err == nil {
 		query.BodyValid = true
 		query.ToolHistoryValid = openAIReasoningToolHistoryAllowsRecovery(body)
 		for _, item := range openAIReasoningCipherItems(body) {
 			query.CipherIndices = append(query.CipherIndices, item.index)
+			if !query.NamedCipher && rejection.message != "" && openAIRecoveryMessageNamesItem(rejection.message, item.id) {
+				query.NamedCipher = true
+			}
 		}
 		conversation := gjson.GetBytes(body, "conversation")
 		query.ServerContext = gjson.GetBytes(body, "previous_response_id").String() != "" || (conversation.Exists() && conversation.Type != gjson.Null)