From 4e2ae09c8ac91c132e1dddc8fa1523446225e38b Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 7 Oct 2026 11:15:41 +0800 Subject: [PATCH 1/3] fix(reasoning): re-send without the id when the upstream cannot find a stripped reasoning item Reasoning recovery deletes encrypted_content from reasoning items and keeps the item with its id. An upstream may take such an item for a reference to one it stores, and answers `Item with id '' not found` when it has none (HTTP 404 from api.openai.com). In production on 2026-10-06 one conversation ended this way 59 times: 56 first sends that had been stripped from the 24h memory and 3 stripped retries, all naming one id of the form rs_ + 32 hex. Ids minted by OpenAI were accepted when left behind in the same way. When the upstream's error message names an item this attempt stripped, the request is now re-sent once on the same account with the id removed from the items the attempt stripped; the items and every other field stay. This is the remedy the Codex OAuth path already applies to every reasoning item. - Trigger: an answer with a status >= 400 other than 401/402/403/407/429 whose message names, as not found, the id of a reasoning item that this attempt stripped and that was sent with that id and without ciphertext. An item without a summary keeps its id, because the upstream refuses it without one. - It is separate from the stripped retry and does not spend it: an account sees at most three POSTs (first send, stripped retry, id repair). A strip made after the repair removes the id together with the ciphertext. - Nothing is remembered. Each request that meets the answer repairs again. - Native Responses, passthrough and Responses-shaped bodies on the Chat Completions endpoint. The repaired answer has no account-health effect. - The wire-to-clean projection accepts the id deletion and nothing else new. - Ops: recovery action `retry_without_item_ids`; the continuation diagnostic gains `recovery.item_ids_removed`. Whether api.openai.com accepts a reasoning item with neither id nor ciphertext was not verified against the real upstream. If it does not, the repaired request ends with that answer instead of the 404. Co-Authored-By: Claude Opus 5.5 --- .downstream/gpt6-models-and-prompts.md | 2 +- ...enai_continuation_account_mismatch_test.go | 25 + .../service/openai_continuation_diagnostic.go | 8 +- .../openai_gateway_chat_completions.go | 12 +- .../service/openai_gateway_forward.go | 15 +- .../service/openai_gateway_passthrough.go | 13 +- .../service/openai_reasoning_recovery.go | 147 +++++- .../openai_reasoning_recovery_item_id_test.go | 488 ++++++++++++++++++ .../internal/service/system_prompt_openai.go | 63 ++- 9 files changed, 744 insertions(+), 29 deletions(-) create mode 100644 backend/internal/service/openai_reasoning_recovery_item_id_test.go diff --git a/.downstream/gpt6-models-and-prompts.md b/.downstream/gpt6-models-and-prompts.md index 8919caceaa79..bdeadfd750ca 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 预算。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出因密文读不了而交给下一个账号的尝试。恢复重发因账号自身的原因失败时也换号（同样只在 `POST /v1/responses`，含 compact 与自动透传；Chat Completions 转 Responses 不换号）：重发已发出、失败属于该路径首次发送时就会换号的那一类，且客户端尚未收到本次尝试的任何输出时，请求按普通账号失败交给下一个账号——下一个账号收到原样请求，本账号不再重试，换号预算同普通换号。哪些失败算这一类以各路径首次发送的换号规则为准：非透传包括额度用尽、限流、鉴权失败、5xx 与流内的上游失败；自动透传的规则更窄，OAuth / Setup Token 透传的普通 5xx、不带账号状态信息的 401/403 等不在其内，仍在本账号结束。Ops 续接诊断用 `recovery.account_failure` 标出这样交给下一个账号的尝试。账号健康：失败的分类与首次发送相同（请求预算被拒 402/403 `balance_insufficient` 只算本次请求，结构化的 `model_not_supported` 只算该账号与模型，`budget_exceeded` 429 算账号，其余池模式可重试的状态不计入 API Key 健康熔断），调度失败照常上报，流内失败照常写入账号状态；重发的 HTTP 错误应答则不执行转发层在首次发送时的账号状态写入（标记限流、置为错误、停用、预算用尽标记、API Key 的瞬时失败计数），也不开启 OAuth 429 的同账号重试窗口，留给之后的首次发送处理，其中首次发送因同时停用账号而把池模式可重试状态计入账号的情形按不计入处理。处理器侧同账号重试用尽后的冷却照常调用，它只对 OAuth / Setup Token 账号起作用（硬额度 429 会写入额度状态）；OpenAI API Key 账号在 HTTP 错误应答下不被冷却，下一个请求仍可能先发给它，由那次首次发送写入状态。重发遇到请求校验错误、拒绝类结果、传输错误、响应头超时或未归类为可换号的错误时，仍在本账号结束；重发再次被报告密文无法解密时，按上文的密文规则换号。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 预算。去掉密文后留下的推理项仍带着 `id`，官方上游可能把它当作对已存储条目的引用去查，查不到时返回 `Item with id '' not found`（官方为 HTTP 404；生产上只见于 `rs_` 加 32 位十六进制的 id，推断由中转签发，OpenAI 自己签发的 id 留着被接受）。上游的错误消息这样点名、且被点名的正是本次尝试去掉过密文的推理项时，把本次尝试去掉过密文的推理项的 `id` 一并去掉（条目和其余字段保留；没有 `summary` 的条目去掉 `id` 后会被上游拒收，所以它保留 `id`，被点名的是它时不修），在同一账号再发一次。这一次与上面的去密文重发各算各的、各自最多一次，所以同一账号最多发三次（首次、去密文、去 id）；先去 id 再遇到密文被拒时，仍可去密文重发一次，届时被去密文的条目同时去掉 `id`。状态为 401/402/403/407/429 时不修；其他 4xx/5xx 只要消息点名了这样的条目就修（中转可能改写状态码）。流内失败不触发。这一步不写任何记忆：同一会话之后的每个请求仍先按原样发出、被拒后再修一次，即每个受影响的请求多一次很快被拒的上传。被修掉的那个错误应答不影响账号健康；修正后的请求若仍失败，按它所处的位置处理（未做过去密文重发时按首次发送，做过时按去密文重发的规则）。原生 Responses、自动透传以及 Chat Completions 端点上的 Responses 形状请求都适用。Ops 记为 `retry_without_item_ids`，续接诊断的 `recovery.item_ids_removed` 是被去掉 `id` 的条目数，`disposition` 仍只描述去密文的那一次重发。官方 API 是否接受不带 `id` 的无密文推理项未用真实上游验证过（Codex OAuth 后端已验证接受）；不接受时修正后的请求以新的错误结束，结果不比修之前差。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出因密文读不了而交给下一个账号的尝试。恢复重发因账号自身的原因失败时也换号（同样只在 `POST /v1/responses`，含 compact 与自动透传；Chat Completions 转 Responses 不换号）：重发已发出、失败属于该路径首次发送时就会换号的那一类，且客户端尚未收到本次尝试的任何输出时，请求按普通账号失败交给下一个账号——下一个账号收到原样请求，本账号不再重试，换号预算同普通换号。哪些失败算这一类以各路径首次发送的换号规则为准：非透传包括额度用尽、限流、鉴权失败、5xx 与流内的上游失败；自动透传的规则更窄，OAuth / Setup Token 透传的普通 5xx、不带账号状态信息的 401/403 等不在其内，仍在本账号结束。Ops 续接诊断用 `recovery.account_failure` 标出这样交给下一个账号的尝试。账号健康：失败的分类与首次发送相同（请求预算被拒 402/403 `balance_insufficient` 只算本次请求，结构化的 `model_not_supported` 只算该账号与模型，`budget_exceeded` 429 算账号，其余池模式可重试的状态不计入 API Key 健康熔断），调度失败照常上报，流内失败照常写入账号状态；重发的 HTTP 错误应答则不执行转发层在首次发送时的账号状态写入（标记限流、置为错误、停用、预算用尽标记、API Key 的瞬时失败计数），也不开启 OAuth 429 的同账号重试窗口，留给之后的首次发送处理，其中首次发送因同时停用账号而把池模式可重试状态计入账号的情形按不计入处理。处理器侧同账号重试用尽后的冷却照常调用，它只对 OAuth / Setup Token 账号起作用（硬额度 429 会写入额度状态）；OpenAI API Key 账号在 HTTP 错误应答下不被冷却，下一个请求仍可能先发给它，由那次首次发送写入状态。重发遇到请求校验错误、拒绝类结果、传输错误、响应头超时或未归类为可换号的错误时，仍在本账号结束；重发再次被报告密文无法解密时，按上文的密文规则换号。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 index fec991336bce..14fa9f60f0e9 100644 --- a/backend/internal/handler/openai_continuation_account_mismatch_test.go +++ b/backend/internal/handler/openai_continuation_account_mismatch_test.go @@ -387,3 +387,28 @@ func TestOpenAIGatewayHandler_ContinuationStrippedRetryAccountFailureCoolsTheAcc serveContinuationMismatchSession(t, handler, body, "another-conversation") require.Equal(t, []int64{1, 1, 2, 2}, upstream.accountIDs, "account 1 would accept again, but is cooling") } + +// The upstream cannot find a reasoning item that the stripped retry left behind +// without ciphertext. The same account is asked once more, without the id, and +// answers; no other account is involved and the session stays where it was. +func TestOpenAIGatewayHandler_ContinuationStrippedItemTheUpstreamCannotFindIsRepairedOnTheSameAccount(t *testing.T) { + gin.SetMode(gin.TestMode) + const relayID = "rs_0123456789abcdef0123456789abcdef" + body := strings.Replace(continuationMismatchReasoningOnlyBody, `"rs_a"`, `"`+relayID+`"`, 1) + unfound := continuationMismatchReply{status: http.StatusNotFound, + payload: `{"error":{"message":"Item with id '` + relayID + `' not found. Items are not persisted when ` + "`store`" + ` is set to false.","type":"invalid_request_error","param":"input","code":null}}`} + upstream := &continuationMismatchUpstream{ + accepting: map[int64]bool{1: true, 2: true}, + sequence: map[int64][]continuationMismatchReply{1: {continuationMismatchUnrecoverable, unfound}}, + } + handler, cache := newContinuationMismatchHandler(t, upstream, continuationMismatchAccounts(2, service.AccountTypeAPIKey)) + response := serveContinuationMismatch(t, handler, body) + + require.Equal(t, http.StatusOK, response.code, response.body) + require.Equal(t, []int64{1, 1, 1}, upstream.accountIDs, "first send, stripped retry, then the repair") + require.Equal(t, relayID, gjson.GetBytes(upstream.bodies[1], "input.0.id").String()) + require.False(t, gjson.GetBytes(upstream.bodies[1], "input.0.encrypted_content").Exists()) + require.Equal(t, "reasoning", gjson.GetBytes(upstream.bodies[2], "input.0.type").String(), "the item is kept") + require.False(t, gjson.GetBytes(upstream.bodies[2], "input.0.id").Exists(), "without the id the upstream could not find") + require.Equal(t, int64(1), continuationMismatchBoundAccount(t, cache)) +} diff --git a/backend/internal/service/openai_continuation_diagnostic.go b/backend/internal/service/openai_continuation_diagnostic.go index 2c209010031d..772f1be5753e 100644 --- a/backend/internal/service/openai_continuation_diagnostic.go +++ b/backend/internal/service/openai_continuation_diagnostic.go @@ -43,7 +43,10 @@ type OpenAIContinuationDiagnostic struct { } type openAIContinuationRecoveryShape struct { - CacheSkippedItems int `json:"cache_skipped_items"` + CacheSkippedItems int `json:"cache_skipped_items"` + // ItemIDsRemoved counts stripped reasoning items sent without their id, + // because the upstream did not find such an id. + ItemIDsRemoved int `json:"item_ids_removed,omitempty"` RetryAttempted bool `json:"retry_attempted"` Disposition string `json:"disposition"` NotAttemptedReason string `json:"not_attempted_reason,omitempty"` @@ -462,6 +465,9 @@ func sanitizeOpenAIContinuationDiagnosticValues(in *OpenAIContinuationDiagnostic if recovery.CacheSkippedItems < 0 { recovery.CacheSkippedItems = 0 } + if recovery.ItemIDsRemoved < 0 { + recovery.ItemIDsRemoved = 0 + } switch recovery.Disposition { case "not_attempted", "retry_prepared", "retry_attempted", "budget_exhausted": default: diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 834eb151bbf9..27df3b89d0d9 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -453,9 +453,17 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions( } if resp.StatusCode >= 400 { respBody, upstreamMsg := s.readOpenAIUpstreamError(resp, c) + if repaired, repair := recovery.TryRepairUnfoundItemIDs(resp.StatusCode, resp.Header, respBody); repair { + closeUpstreamResponse() + responsesBody, err = projectReasoningRecoveryEdits(responsesBody, wireBody, repaired) + if err != nil { + return nil, err + } + continue + } if retryBody, retry := recovery.TryRecover(resp.StatusCode, resp.Header, respBody, false); retry { closeUpstreamResponse() - responsesBody, err = projectReasoningCipherEdits(responsesBody, wireBody, retryBody) + responsesBody, err = projectReasoningRecoveryEdits(responsesBody, wireBody, retryBody) if err != nil { return nil, err } @@ -495,7 +503,7 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions( } closeUpstreamResponse() if retryBody, retry := recovery.TryRecoverError(handleErr); retry { - responsesBody, err = projectReasoningCipherEdits(responsesBody, wireBody, retryBody) + responsesBody, err = projectReasoningRecoveryEdits(responsesBody, wireBody, retryBody) if err != nil { return nil, err } diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index ca7ec69c16a1..2df9ae8c5774 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1076,8 +1076,17 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco respBody := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() resp.Body = io.NopCloser(bytes.NewReader(respBody)) + if repaired, repair := reasoningRecovery.TryRepairUnfoundItemIDs(resp.StatusCode, resp.Header, respBody); repair { + body, err = projectReasoningRecoveryEdits(body, wireBody, repaired) + if err != nil { + return nil, err + } + requestView = newOpenAIRequestView(body) + reqBody = nil + continue + } if retryBody, retry := reasoningRecovery.TryRecover(resp.StatusCode, resp.Header, respBody, false); retry { - body, err = projectReasoningCipherEdits(body, wireBody, retryBody) + body, err = projectReasoningRecoveryEdits(body, wireBody, retryBody) if err != nil { return nil, err } @@ -1223,7 +1232,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if err != nil { if retryBody, retry := reasoningRecovery.TryRecoverError(err); retry { _ = resp.Body.Close() - body, err = projectReasoningCipherEdits(body, wireBody, retryBody) + body, err = projectReasoningRecoveryEdits(body, wireBody, retryBody) if err != nil { return nil, err } @@ -1285,7 +1294,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if err != nil { if retryBody, retry := reasoningRecovery.TryRecoverError(err); retry { _ = resp.Body.Close() - body, err = projectReasoningCipherEdits(body, wireBody, retryBody) + body, err = projectReasoningRecoveryEdits(body, wireBody, retryBody) if err != nil { return nil, err } diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 7e12f364e6ad..925b10deb694 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -429,8 +429,15 @@ retryUpstream: probeBody := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() resp.Body = io.NopCloser(bytes.NewReader(probeBody)) + if repaired, repair := reasoningRecovery.TryRepairUnfoundItemIDs(resp.StatusCode, resp.Header, probeBody); repair { + body, err = projectReasoningRecoveryEdits(body, wireBody, repaired) + if err != nil { + return nil, err + } + continue + } if retryBody, retry := reasoningRecovery.TryRecover(resp.StatusCode, resp.Header, probeBody, false); retry { - body, err = projectReasoningCipherEdits(body, wireBody, retryBody) + body, err = projectReasoningRecoveryEdits(body, wireBody, retryBody) if err != nil { return nil, err } @@ -531,7 +538,7 @@ retryUpstream: if err != nil { if retryBody, retry := reasoningRecovery.TryRecoverError(err); retry { _ = resp.Body.Close() - body, err = projectReasoningCipherEdits(body, wireBody, retryBody) + body, err = projectReasoningRecoveryEdits(body, wireBody, retryBody) if err != nil { return nil, err } @@ -568,7 +575,7 @@ retryUpstream: if err != nil { if retryBody, retry := reasoningRecovery.TryRecoverError(err); retry { _ = resp.Body.Close() - body, err = projectReasoningCipherEdits(body, wireBody, retryBody) + body, err = projectReasoningRecoveryEdits(body, wireBody, retryBody) if err != nil { return nil, err } diff --git a/backend/internal/service/openai_reasoning_recovery.go b/backend/internal/service/openai_reasoning_recovery.go index 68a8c66d8ce8..05919b2677ed 100644 --- a/backend/internal/service/openai_reasoning_recovery.go +++ b/backend/internal/service/openai_reasoning_recovery.go @@ -43,6 +43,10 @@ const ( // 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. +// +// Apart from that retry it may re-send once more to undo a side effect of its +// own strip: an upstream that looks up by id an item left without ciphertext +// and does not find it (TryRepairUnfoundItemIDs). type openAIReasoningRecoveryState struct { ctx context.Context c *gin.Context @@ -86,6 +90,16 @@ type openAIReasoningRecoveryState struct { // accountFailure records that the stripped retry failed for a reason of // the account's own. accountFailure bool + + // stripped lists the reasoning items whose ciphertext this attempt removed, + // by their position in the wire body. + stripped []openAIReasoningCipherItem + // dropItemIDs makes a strip remove the item's id with its ciphertext. It + // is set by the repair, for the rest of this attempt. + dropItemIDs bool + // idRepairUsed records the one re-send without the ids of stripped items. + idRepairUsed bool + itemIDsRemoved int } func (s *OpenAIGatewayService) newOpenAIReasoningRecoveryState(ctx context.Context, c *gin.Context, account *Account, token string) *openAIReasoningRecoveryState { @@ -360,25 +374,44 @@ func (r *openAIReasoningRecoveryState) skipRejectedHistory(body []byte) []byte { return body } now := time.Now() - indices := make([]int, 0, len(items)) + skipped := make([]openAIReasoningCipherItem, 0, len(items)) for _, item := range items { if old, ok := rejected[item.hash]; ok && !old.RejectedAt.IsZero() && !old.RejectedAt.After(now) && now.Before(old.ExpiresAt) && now.Before(old.RejectedAt.Add(OpenAIReasoningStateTTL)) { - indices = append(indices, item.index) + skipped = append(skipped, item) } } - if len(indices) == 0 { + if len(skipped) == 0 { return body } - stripped, err := stripOpenAIReasoningCipherIndices(body, indices) + stripped, err := r.strip(body, skipped) if err != nil { return body } - r.cacheSkippedItems += len(indices) - r.record("rejected_history_skipped", 0, "", nil, len(indices)) + r.cacheSkippedItems += len(skipped) + r.record("rejected_history_skipped", 0, "", nil, len(skipped)) return stripped } +// strip removes the ciphertext of items from body and remembers them as +// stripped by this attempt. After the repair the id goes with the ciphertext. +func (r *openAIReasoningRecoveryState) strip(body []byte, items []openAIReasoningCipherItem) ([]byte, error) { + edit := openAIReasoningRecoveryEdit{} + for _, item := range items { + edit.cipher = append(edit.cipher, item.index) + if r.dropItemIDs && item.id != "" && openAIReasoningItemStandsWithoutID(gjson.GetBytes(body, fmt.Sprintf("input.%d", item.index))) { + edit.ids = append(edit.ids, item.index) + } + } + stripped, err := applyOpenAIReasoningRecoveryEdit(body, edit) + if err != nil { + return nil, err + } + r.stripped = append(r.stripped, items...) + r.itemIDsRemoved += len(edit.ids) + return stripped, nil +} + type openAIReasoningCipherItem struct { index int hash string @@ -498,6 +531,10 @@ func countOpenAIEncryptedFields(value any) int { } func stripOpenAIReasoningCipherIndices(body []byte, indices []int) ([]byte, error) { + return deleteOpenAIReasoningItemField(body, indices, "encrypted_content") +} + +func deleteOpenAIReasoningItemField(body []byte, indices []int, field string) ([]byte, error) { out := bytes.Clone(body) for _, index := range indices { path := fmt.Sprintf("input.%d", index) @@ -505,7 +542,7 @@ func stripOpenAIReasoningCipherIndices(body []byte, indices []int) ([]byte, erro return nil, errors.New("reasoning recovery target mismatch") } var err error - out, err = sjson.DeleteBytes(out, path+".encrypted_content") + out, err = sjson.DeleteBytes(out, path+"."+field) if err != nil { return nil, err } @@ -513,6 +550,34 @@ func stripOpenAIReasoningCipherIndices(body []byte, indices []int) ([]byte, erro return out, nil } +// openAIReasoningRecoveryEdit is everything recovery may do to a request: at +// the listed input positions, all reasoning items, delete encrypted_content +// and delete id. Positions never move. +type openAIReasoningRecoveryEdit struct{ cipher, ids []int } + +// applyOpenAIReasoningRecoveryEdit makes the edit in one fixed order, so that +// the same edit yields the same bytes wherever it is repeated. +func applyOpenAIReasoningRecoveryEdit(body []byte, edit openAIReasoningRecoveryEdit) ([]byte, error) { + out, err := deleteOpenAIReasoningItemField(body, edit.cipher, "encrypted_content") + if err != nil { + return nil, err + } + return deleteOpenAIReasoningItemField(out, edit.ids, "id") +} + +// openAIReasoningItemHasCipher is the test openAIReasoningCipherItems applies. +func openAIReasoningItemHasCipher(item gjson.Result) bool { + cipher := item.Get("encrypted_content") + return cipher.Type == gjson.String && cipher.String() != "" +} + +// openAIReasoningItemStandsWithoutID reports an item the upstream takes as it +// is once its id is gone. An item without a summary is refused then, so such +// an item keeps its id. +func openAIReasoningItemStandsWithoutID(item gjson.Result) bool { + return item.Get("summary").IsArray() +} + // Signal is deliberately pure. A bare SSE error may be superseded by a later // authoritative failed/completed event and must not consume retry or cache state. func openAIReasoningRecoverySignal(c *gin.Context, payload []byte, semanticCommitted bool) error { @@ -567,7 +632,16 @@ func (r *openAIReasoningRecoveryState) TryRecover(status int, headers http.Heade if len(indices) == 0 { return nil, false } - stripped, err := stripOpenAIReasoningCipherIndices(r.wire, indices) + selected := make([]openAIReasoningCipherItem, 0, len(indices)) + for _, item := range openAIReasoningCipherItems(r.wire) { + if slices.Contains(indices, item.index) { + selected = append(selected, item) + } + } + if len(selected) != len(indices) { + return nil, false + } + stripped, err := r.strip(r.wire, selected) if err != nil { return nil, false } @@ -588,6 +662,62 @@ func (r *openAIReasoningRecoveryState) TryRecover(status int, headers http.Heade return stripped, true } +// TryRepairUnfoundItemIDs undoes a side effect of this attempt's own strip. A +// reasoning item left without ciphertext still carries its id, and an upstream +// may take such an item for a reference to one it stores. When it does not +// find it, it says so and names the id (officially with status 404); without +// the id it takes the item as it is. The repair is made only for that answer +// naming an item this attempt stripped, removes the id of every such item, and +// is re-sent once. It is not the stripped retry and does not spend it. +// +// Nothing is remembered: every attempt that meets the answer repairs again. +func (r *openAIReasoningRecoveryState) TryRepairUnfoundItemIDs(status int, headers http.Header, payload []byte) ([]byte, bool) { + if r == nil || !r.enabled || r.idRepairUsed || len(r.stripped) == 0 || r.ctx.Err() != nil || + status < http.StatusBadRequest || openAIReasoningRecoveryProtectedStatus(status) { + return nil, false + } + // The official answer to an item looked up by id and not found. A relay may + // pass it on under another status; the named id is what identifies it. + message := extractUpstreamErrorMessage(payload) + if !strings.Contains(message, "Item with id '") { + return nil, false + } + found := false + ids := make([]int, 0, len(r.stripped)) + for _, item := range r.stripped { + // What was stripped is checked against what was sent. + sent := gjson.GetBytes(r.wire, fmt.Sprintf("input.%d", item.index)) + id := sent.Get("id") + if item.id == "" || sent.Get("type").String() != "reasoning" || openAIReasoningItemHasCipher(sent) || + id.Type != gjson.String || id.String() != item.id || !openAIReasoningItemStandsWithoutID(sent) || + slices.Contains(ids, item.index) { + continue + } + ids = append(ids, item.index) + found = found || strings.Contains(message, "Item with id '"+item.id+"' not found") + } + if !found { + return nil, false + } + repaired, err := applyOpenAIReasoningRecoveryEdit(r.wire, openAIReasoningRecoveryEdit{ids: ids}) + if err != nil { + return nil, false + } + r.idRepairUsed, r.dropItemIDs = true, true + r.itemIDsRemoved += len(ids) + if r.retryUsed { + // The stripped retry was already sent; the send boundary accepts only + // the body it knows, and this one is not sent yet. + r.retryBody = bytes.Clone(repaired) + r.retryDispatched, r.diagnosticState = false, "retry_prepared" + } + if r.responseStatus == 0 { + r.responseStatus, r.responseHeaders = status, headers.Clone() + } + r.record("retry_without_item_ids", status, "", payload, len(ids)) + return repaired, true +} + func openAIReasoningRecoveryProtectedStatus(status int) bool { switch status { case http.StatusUnauthorized, http.StatusPaymentRequired, http.StatusForbidden, http.StatusProxyAuthRequired, http.StatusTooManyRequests: @@ -800,6 +930,7 @@ func (r *openAIReasoningRecoveryState) requestRejectionFromStream(payload []byte func (r *openAIReasoningRecoveryState) continuationDiagnosticRecovery(payload []byte) *openAIContinuationRecoveryShape { shape := &openAIContinuationRecoveryShape{ CacheSkippedItems: r.cacheSkippedItems, + ItemIDsRemoved: r.itemIDsRemoved, RetryAttempted: r.retryDispatched, Disposition: r.diagnosticState, AccountMismatch: r.accountMismatch, diff --git a/backend/internal/service/openai_reasoning_recovery_item_id_test.go b/backend/internal/service/openai_reasoning_recovery_item_id_test.go new file mode 100644 index 000000000000..91de6655aac8 --- /dev/null +++ b/backend/internal/service/openai_reasoning_recovery_item_id_test.go @@ -0,0 +1,488 @@ +//go:build unit + +package service + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" +) + +// An id in the form a relay mints. The official upstream takes a reasoning +// item that carries such an id and no ciphertext for a reference to an item it +// stores, and does not find it. +const reasoningRecoveryRelayItemID = "rs_0123456789abcdef0123456789abcdef" + +var reasoningRecoveryRelayFixture = strings.Replace(reasoningRecoveryFixture, "rs_old", reasoningRecoveryRelayItemID, 1) + +func reasoningRecoveryJSON(status int, body string) *http.Response { + resp := newJSONResponse(status, body) + resp.Header.Set("Content-Type", "application/json") + return resp +} + +// The official answer, as recorded in production. +func reasoningRecoveryUnfoundItem(id string) *http.Response { + return reasoningRecoveryJSON(http.StatusNotFound, `{"error":{"message":"Item with id '`+id+`' not found. Items are not persisted when `+"`store`"+` is set to false. Try again with `+"`store`"+` set to true, or remove this item from your input.","type":"invalid_request_error","param":"input","code":null}}`) +} + +func reasoningRecoveryCipherRejected() *http.Response { + return reasoningRecoveryJSON(http.StatusBadRequest, `{"error":{"message":"The encrypted content could not be verified.","type":"invalid_request_error","param":null,"code":"invalid_encrypted_content"}}`) +} + +func reasoningRecoveryCompleted() *http.Response { + return reasoningRecoveryJSON(http.StatusOK, `{"id":"resp_final","status":"completed","model":"gpt-5.6-sol","output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":"ok"}]}],"usage":{"input_tokens":11,"output_tokens":7}}`) +} + +// reasoningRecoveryItemIDHarness forwards requests of one key to one account +// over a scripted upstream and a rejection memory that outlives a request. +type reasoningRecoveryItemIDHarness struct { + svc *OpenAIGatewayService + store *reasoningRecoveryMemoryStore + upstream *httpUpstreamRecorder + account *Account +} + +func newReasoningRecoveryItemIDHarness(passthrough bool) *reasoningRecoveryItemIDHarness { + upstream := &httpUpstreamRecorder{} + h := &reasoningRecoveryItemIDHarness{ + svc: newOpenAIImageGenerationControlTestService(upstream), upstream: upstream, + store: &reasoningRecoveryMemoryStore{GatewayCache: &stubGatewayCache{}}, account: newOpenAIImageGenerationControlTestAccount(), + } + h.svc.cache = h.store + h.account.Extra = map[string]any{"openai_passthrough": passthrough, "responses_api_supported": true} + return h +} + +type reasoningRecoveryItemIDAttempt struct { + c *gin.Context + recorder *httptest.ResponseRecorder + bodies [][]byte + err error +} + +// forward sends body once; the upstream answers with responses in order. +func (h *reasoningRecoveryItemIDHarness) forward(t *testing.T, body string, responses ...*http.Response) reasoningRecoveryItemIDAttempt { + t.Helper() + c, recorder := newOpenAIImageGenerationControlTestContext(false, "test-client") + getAPIKeyFromContext(c).UserID = 77 + sent := len(h.upstream.bodies) + h.upstream.responses = responses + _, err := h.svc.Forward(context.Background(), c, h.account, []byte(body)) + require.Empty(t, h.upstream.responses, "every scripted answer was asked for") + return reasoningRecoveryItemIDAttempt{c: c, recorder: recorder, bodies: h.upstream.bodies[sent:], err: err} +} + +// learn makes the memory hold the ciphertext of body's reasoning items: the +// upstream rejects it once and accepts the stripped request. +func (h *reasoningRecoveryItemIDHarness) learn(t *testing.T, body string) { + t.Helper() + attempt := h.forward(t, body, reasoningRecoveryCipherRejected(), reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 2) + require.Equal(t, 1, h.store.puts) +} + +func (a reasoningRecoveryItemIDAttempt) recoveryActions() []string { + events, _ := a.c.Get(OpsUpstreamErrorsKey) + var actions []string + recorded, _ := events.([]*OpsUpstreamErrorEvent) + for _, event := range recorded { + if event.Kind == "reasoning_recovery" { + actions = append(actions, event.Message) + } + } + return actions +} + +func forEachReasoningRecoveryPath(t *testing.T, run func(t *testing.T, h *reasoningRecoveryItemIDHarness)) { + for _, passthrough := range []bool{false, true} { + t.Run(map[bool]string{true: "passthrough", false: "native"}[passthrough], func(t *testing.T) { + run(t, newReasoningRecoveryItemIDHarness(passthrough)) + }) + } +} + +// The upstream cannot find an item the gateway left behind without ciphertext. +// The request is sent once more with the ids of the stripped items removed and +// nothing else changed. +func TestOpenAIReasoningRecoveryRepairsAStrippedItemTheUpstreamCannotFind(t *testing.T) { + t.Run("pre_stripped_first_send", func(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + h.learn(t, reasoningRecoveryRelayFixture) + + attempt := h.forward(t, reasoningRecoveryRelayFixture, reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 2, "the pre-stripped send and the repair") + require.False(t, gjson.GetBytes(attempt.bodies[0], "input.1.encrypted_content").Exists()) + require.Equal(t, reasoningRecoveryRelayItemID, gjson.GetBytes(attempt.bodies[0], "input.1.id").String()) + withoutID, err := sjson.DeleteBytes(attempt.bodies[0], "input.1.id") + require.NoError(t, err) + require.Equal(t, string(withoutID), string(attempt.bodies[1]), "only the id of the stripped item is gone") + require.Equal(t, "visible summary", gjson.GetBytes(attempt.bodies[1], "input.1.summary.0.text").String(), "the item itself stays") + require.Equal(t, "fc_one", gjson.GetBytes(attempt.bodies[1], "input.2.id").String(), "no other item loses its id") + require.Contains(t, attempt.recorder.Body.String(), "resp_final") + require.Equal(t, []string{"rejected_history_skipped", "retry_without_item_ids"}, attempt.recoveryActions()) + require.Equal(t, "retry_without_item_ids", attempt.recorder.Header().Get(openAIReasoningRecoveryHeader)) + require.Equal(t, 1, h.store.puts, "the repair writes nothing to the memory") + }) + }) + t.Run("stripped_retry", func(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + attempt := h.forward(t, reasoningRecoveryRelayFixture, + reasoningRecoveryCipherRejected(), reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 3, "the first send, the stripped retry and the repair") + require.Equal(t, "opaque-old", gjson.GetBytes(attempt.bodies[0], "input.1.encrypted_content").String()) + stripped, err := sjson.DeleteBytes(attempt.bodies[0], "input.1.encrypted_content") + require.NoError(t, err) + require.Equal(t, string(stripped), string(attempt.bodies[1])) + withoutID, err := sjson.DeleteBytes(stripped, "input.1.id") + require.NoError(t, err) + require.Equal(t, string(withoutID), string(attempt.bodies[2])) + require.Contains(t, attempt.recorder.Body.String(), "resp_final") + require.Equal(t, []string{"retry_without_encrypted_content", "retry_without_item_ids"}, attempt.recoveryActions()) + require.Equal(t, 1, h.store.puts, "the rejected ciphertext only") + }) + }) + // A relay may pass the official answer on under a status of its own. + t.Run("answered_with_another_status", func(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + h.learn(t, reasoningRecoveryRelayFixture) + wrapped := reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID) + wrapped.StatusCode = http.StatusBadRequest + + attempt := h.forward(t, reasoningRecoveryRelayFixture, wrapped, reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 2) + require.False(t, gjson.GetBytes(attempt.bodies[1], "input.1.id").Exists()) + }) + }) +} + +// Nothing about the repair is remembered: the next request of the conversation +// is sent as before, meets the same answer and is repaired again. +func TestOpenAIReasoningRecoveryRepairIsMadeAgainOnTheNextRequest(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + h.learn(t, reasoningRecoveryRelayFixture) + first := h.forward(t, reasoningRecoveryRelayFixture, reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCompleted()) + require.NoError(t, first.err) + + next := h.forward(t, reasoningRecoveryRelayFixture, reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCompleted()) + require.NoError(t, next.err) + require.Len(t, next.bodies, 2) + require.Equal(t, string(first.bodies[0]), string(next.bodies[0]), "the id is still sent first") + require.Equal(t, string(first.bodies[1]), string(next.bodies[1])) + }) +} + +// The upstream names the first item it cannot find. Every stripped item loses +// its id in the one repair, so the next one is not reported in turn. +func TestOpenAIReasoningRecoveryRepairRemovesTheIDOfEveryStrippedItem(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + body := strings.Replace(reasoningRecoveryRelayFixture, `"output":"[]"}`, + `"output":"[]"},{"type":"reasoning","id":"rs_fedcba9876543210fedcba9876543210","summary":[],"encrypted_content":"opaque-second"}`, 1) + h.learn(t, body) + + attempt := h.forward(t, body, reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 2) + for _, index := range []string{"1", "4"} { + require.True(t, gjson.GetBytes(attempt.bodies[0], "input."+index+".id").Exists()) + require.False(t, gjson.GetBytes(attempt.bodies[0], "input."+index+".encrypted_content").Exists()) + require.Equal(t, "reasoning", gjson.GetBytes(attempt.bodies[1], "input."+index+".type").String()) + require.False(t, gjson.GetBytes(attempt.bodies[1], "input."+index+".id").Exists()) + } + }) +} + +// A repair that happens before any stripped retry leaves that retry available: +// ciphertext the upstream then rejects is still stripped once, and its item +// loses the id with it. +func TestOpenAIReasoningRecoveryRepairLeavesTheStrippedRetryAvailable(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + h.learn(t, reasoningRecoveryRelayFixture) + body := strings.Replace(reasoningRecoveryRelayFixture, `"output":"[]"}`, + `"output":"[]"},{"type":"reasoning","id":"rs_fedcba9876543210fedcba9876543210","summary":[],"encrypted_content":"opaque-new"}`, 1) + + attempt := h.forward(t, body, + reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCipherRejected(), reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 3) + require.Equal(t, "opaque-new", gjson.GetBytes(attempt.bodies[1], "input.4.encrypted_content").String()) + require.False(t, gjson.GetBytes(attempt.bodies[1], "input.1.id").Exists()) + for _, index := range []string{"1", "4"} { + require.Equal(t, "reasoning", gjson.GetBytes(attempt.bodies[2], "input."+index+".type").String()) + require.False(t, gjson.GetBytes(attempt.bodies[2], "input."+index+".encrypted_content").Exists()) + require.False(t, gjson.GetBytes(attempt.bodies[2], "input."+index+".id").Exists(), "after the repair a strip takes the id too") + } + require.Equal(t, []string{"rejected_history_skipped", "retry_without_item_ids", "retry_without_encrypted_content"}, attempt.recoveryActions()) + }) +} + +// The repair is made once in an attempt. +func TestOpenAIReasoningRecoveryRepairsOnce(t *testing.T) { + validation := func() *http.Response { + return reasoningRecoveryJSON(http.StatusBadRequest, `{"error":{"message":"Missing required parameter: 'input[1].id'.","type":"invalid_request_error","param":"input[1].id","code":"missing_required_parameter"}}`) + } + for _, test := range []struct { + name string + answer func() *http.Response + }{ + {name: "not_found_again", answer: func() *http.Response { return reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID) }}, + {name: "rejected", answer: validation}, + } { + t.Run(test.name, func(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + h.learn(t, reasoningRecoveryRelayFixture) + + attempt := h.forward(t, reasoningRecoveryRelayFixture, reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), test.answer()) + require.Error(t, attempt.err) + require.Len(t, attempt.bodies, 2, "no second repair") + require.Equal(t, 1, h.store.puts) + }) + }) + } + t.Run("after_the_stripped_retry", func(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + attempt := h.forward(t, reasoningRecoveryRelayFixture, reasoningRecoveryCipherRejected(), + reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID)) + require.Len(t, attempt.bodies, 3) + var stopped *OpenAIReasoningRecoveryTerminalError + require.ErrorAs(t, attempt.err, &stopped, "the stripped retry is spent: the request ends on this account") + require.Equal(t, http.StatusNotFound, stopped.Failure.StatusCode) + recovery := lastReasoningRecoveryDiagnostic(t, attempt.c).ContinuationDiagnostic.Recovery + require.Equal(t, 1, recovery.ItemIDsRemoved) + require.True(t, recovery.RetryAttempted, "the repaired body was sent") + }) + }) +} + +// Only the answer naming an item this attempt stripped is repaired. +func TestOpenAIReasoningRecoveryLeavesEveryOtherMissingItemAlone(t *testing.T) { + const clientItemID = "rs_ffffffffffffffffffffffffffffffff" + withClientItem := strings.Replace(reasoningRecoveryRelayFixture, `"output":"[]"}`, + `"output":"[]"},{"type":"reasoning","id":"`+clientItemID+`","summary":[]}`, 1) + withoutSummary := strings.Replace(reasoningRecoveryRelayFixture, `"summary":[{"type":"summary_text","text":"visible summary"}],`, ``, 1) + for _, test := range []struct { + name string + learn string + body string + answer func() *http.Response + }{ + { + name: "an_item_the_client_sent_without_ciphertext", learn: reasoningRecoveryRelayFixture, body: withClientItem, + answer: func() *http.Response { return reasoningRecoveryUnfoundItem(clientItemID) }, + }, + { + name: "an_id_that_is_not_in_the_request", learn: reasoningRecoveryRelayFixture, body: reasoningRecoveryRelayFixture, + answer: func() *http.Response { return reasoningRecoveryUnfoundItem("rs_00000000000000000000000000000000") }, + }, + { + name: "nothing_was_stripped", body: reasoningRecoveryRelayFixture, + answer: func() *http.Response { return reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID) }, + }, + { + // Without a summary the upstream refuses the item once its id is gone. + name: "an_item_without_a_summary", learn: withoutSummary, body: withoutSummary, + answer: func() *http.Response { return reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID) }, + }, + { + name: "a_status_that_says_something_about_the_account", learn: reasoningRecoveryRelayFixture, body: reasoningRecoveryRelayFixture, + answer: func() *http.Response { + forbidden := reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID) + forbidden.StatusCode = http.StatusForbidden + return forbidden + }, + }, + { + name: "another_message", learn: reasoningRecoveryRelayFixture, body: reasoningRecoveryRelayFixture, + answer: func() *http.Response { + return reasoningRecoveryJSON(http.StatusNotFound, `{"error":{"message":"The model `+"`"+reasoningRecoveryRelayItemID+"`"+` does not exist.","type":"invalid_request_error","param":"input","code":null}}`) + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + if test.learn != "" { + h.learn(t, test.learn) + } + attempt := h.forward(t, test.body, test.answer()) + require.Error(t, attempt.err) + require.Len(t, attempt.bodies, 1, "the request is not sent again") + require.NotContains(t, attempt.recoveryActions(), "retry_without_item_ids") + }) + }) + } +} + +// An item without a summary keeps its id also when a strip follows the repair. +func TestOpenAIReasoningRecoveryKeepsTheIDOfAnItemWithoutSummary(t *testing.T) { + forEachReasoningRecoveryPath(t, func(t *testing.T, h *reasoningRecoveryItemIDHarness) { + h.learn(t, reasoningRecoveryRelayFixture) + const bareID = "rs_fedcba9876543210fedcba9876543210" + body := strings.Replace(reasoningRecoveryRelayFixture, `"output":"[]"}`, + `"output":"[]"},{"type":"reasoning","id":"`+bareID+`","encrypted_content":"opaque-new"}`, 1) + + attempt := h.forward(t, body, + reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), reasoningRecoveryCipherRejected(), reasoningRecoveryCompleted()) + require.NoError(t, attempt.err) + require.Len(t, attempt.bodies, 3) + require.False(t, gjson.GetBytes(attempt.bodies[2], "input.1.id").Exists()) + require.False(t, gjson.GetBytes(attempt.bodies[2], "input.4.encrypted_content").Exists()) + require.Equal(t, bareID, gjson.GetBytes(attempt.bodies[2], "input.4.id").String()) + }) +} + +// The repair acts on the body that was sent: an item it stripped must be there, +// at its place, with its id and without ciphertext. +func TestOpenAIReasoningRecoveryRepairChecksWhatItStrippedAgainstWhatWasSent(t *testing.T) { + unfound := []byte(`{"error":{"message":"Item with id '` + reasoningRecoveryRelayItemID + `' not found."}}`) + prepared := func(t *testing.T) *openAIReasoningRecoveryState { + state, _, _ := newReasoningRecoveryTestState(t, context.Background(), reasoningRecoveryRelayFixture) + _, repaired := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) + require.False(t, repaired, "nothing has been stripped") + _, retry := state.TryRecover(http.StatusBadRequest, nil, []byte(`{"error":{"code":"invalid_encrypted_content"}}`), false) + require.True(t, retry) + return state + } + + t.Run("the_stripped_body_was_not_sent", func(t *testing.T) { + state := prepared(t) + _, repaired := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) + require.False(t, repaired, "the body last sent still holds the ciphertext") + }) + for name, sent := range map[string]func(wire []byte) []byte{ + "another_id_at_its_place": func(wire []byte) []byte { + out, _ := sjson.SetBytes(wire, "input.1.id", "rs_ffffffffffffffffffffffffffffffff") + return out + }, + "another_item_at_its_place": func(wire []byte) []byte { + out, _ := sjson.SetBytes(wire, "input.1.type", "message") + return out + }, + } { + t.Run(name, func(t *testing.T) { + state := prepared(t) + state.wire = sent(state.retryBody) + _, repaired := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) + require.False(t, repaired) + }) + } + t.Run("sent_as_stripped", func(t *testing.T) { + state := prepared(t) + state.MarkAttemptDispatched() + state.wire = bytes.Clone(state.retryBody) + body, repaired := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) + require.True(t, repaired) + require.False(t, gjson.GetBytes(body, "input.1.id").Exists()) + require.Equal(t, string(body), string(state.retryBody), "the send boundary accepts the repaired body") + require.False(t, state.retryDispatched, "which is not sent yet") + _, again := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) + require.False(t, again) + }) +} + +// A Responses-shaped body on the Chat Completions endpoint is stripped by the +// same rules and is repaired the same way. +func TestOpenAIReasoningRecoveryChatPathRepairsAStrippedItemTheUpstreamCannotFind(t *testing.T) { + store := &reasoningRecoveryMemoryStore{GatewayCache: &stubGatewayCache{}} + account := newOpenAIRejectedFieldTestAccount() + fixture := strings.Replace(reasoningRecoveryChatFixture, `{"type":"reasoning",`, `{"type":"reasoning","id":"`+reasoningRecoveryRelayItemID+`",`, 1) + body, err := sjson.SetBytes([]byte(fixture), "stream", true) + require.NoError(t, err) + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + reasoningRecoveryCipherRejected(), + reasoningRecoveryUnfoundItem(reasoningRecoveryRelayItemID), + reasoningRecoverySSEResponse("data: "+reasoningRecoveryChatCompleted+"\n\n", http.StatusOK), + }} + svc := newOpenAIRejectedFieldTestService(upstream) + svc.cache = store + c, _ := reasoningRecoveryChatContext(t, body) + + _, err = svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + require.NoError(t, err) + require.Len(t, upstream.requests, 3, "the first send, the stripped retry and the repair") + require.Equal(t, reasoningRecoveryRelayItemID, gjson.GetBytes(upstream.bodies[1], "input.1.id").String()) + require.Equal(t, "reasoning", gjson.GetBytes(upstream.bodies[2], "input.1.type").String()) + require.False(t, gjson.GetBytes(upstream.bodies[2], "input.1.id").Exists()) + require.False(t, gjson.GetBytes(upstream.bodies[2], "input.1.encrypted_content").Exists()) +} + +// The edit is repeated on the clean body exactly, also where the wire body +// holds an item the clean body does not, and nothing else passes as an edit. +func TestOpenAIReasoningRecoveryProjectionRepeatsOnlyRecoveryEdits(t *testing.T) { + const ( + system = `{"type":"message","role":"developer","content":"system"}` + message = `{"type":"message","id":"msg_1","role":"user","content":"hello"}` + ) + body := func(items ...string) string { return `{"model":"m","input":[` + strings.Join(items, ",") + `]}` } + cipher := func(id, text string) string { + return `{"type":"reasoning","id":"` + id + `","summary":[],"encrypted_content":"` + text + `"}` + } + bare := func(id string) string { return `{"type":"reasoning","id":"` + id + `","summary":[]}` } + const idless = `{"type":"reasoning","summary":[]}` + + for _, test := range []struct { + name, clean, before, after, want string + }{ + { + name: "ciphertext_only", + clean: body(cipher("rs_a", "a"), message), before: body(system, cipher("rs_a", "a"), message), + after: body(system, bare("rs_a"), message), want: body(bare("rs_a"), message), + }, + { + name: "ciphertext_and_id", + clean: body(cipher("rs_a", "a"), cipher("rs_b", "b")), before: body(system, cipher("rs_a", "a"), cipher("rs_b", "b")), + after: body(system, idless, cipher("rs_b", "b")), want: body(idless, cipher("rs_b", "b")), + }, + { + name: "id_of_an_item_left_behind", + clean: body(bare("rs_a"), message, bare("rs_b")), before: body(system, bare("rs_a"), message, bare("rs_b")), + after: body(system, bare("rs_a"), message, idless), want: body(bare("rs_a"), message, idless), + }, + { + name: "second_of_two_items_with_one_id", + clean: body(bare("rs_a"), bare("rs_a")), before: body(system, bare("rs_a"), bare("rs_a")), + after: body(system, bare("rs_a"), idless), want: body(bare("rs_a"), idless), + }, + { + name: "the_id_of_another_item", + clean: body(bare("rs_a"), message), before: body(bare("rs_a"), message), + after: body(bare("rs_a"), `{"type":"message","role":"user","content":"hello"}`), + }, + { + name: "the_id_of_an_item_that_keeps_its_ciphertext", + clean: body(cipher("rs_a", "a")), before: body(cipher("rs_a", "a")), + after: body(`{"type":"reasoning","summary":[],"encrypted_content":"a"}`), + }, + { + name: "any_other_field", + clean: body(bare("rs_a")), before: body(bare("rs_a")), + after: body(`{"type":"reasoning","id":"rs_a"}`), + }, + { + name: "an_item_the_clean_body_does_not_hold", + clean: body(message), before: body(bare("rs_a"), message), + after: body(idless, message), + }, + } { + t.Run(test.name, func(t *testing.T) { + projected, err := projectReasoningRecoveryEdits([]byte(test.clean), []byte(test.before), []byte(test.after)) + if test.want == "" { + require.ErrorIs(t, err, errReasoningWireProjection) + return + } + require.NoError(t, err) + require.Equal(t, test.want, string(projected)) + }) + } +} diff --git a/backend/internal/service/system_prompt_openai.go b/backend/internal/service/system_prompt_openai.go index 2e52e6a53443..54bc88f5a153 100644 --- a/backend/internal/service/system_prompt_openai.go +++ b/backend/internal/service/system_prompt_openai.go @@ -188,7 +188,7 @@ func prepareReasoningRecoveryRequest(recovery *openAIReasoningRecoveryState, req if err != nil { return nil, nil, nil, err } - clean, err = projectReasoningCipherEdits(clean, wire, final) + clean, err = projectReasoningRecoveryEdits(clean, wire, final) if err != nil { return nil, nil, nil, err } @@ -222,11 +222,12 @@ func finalWirePromptCacheKey(wire []byte, seed string) string { return seed } -// projectReasoningCipherEdits repeats a recovery edit on the clean request. +// projectReasoningRecoveryEdits repeats a recovery edit on the clean request. // Recovery may only delete encrypted_content from reasoning items, whose -// relative order the system prompt never changes; any other difference is -// rejected instead of becoming the next retry's source. -func projectReasoningCipherEdits(clean, before, after []byte) ([]byte, error) { +// relative order the system prompt never changes, and the id of an item it +// has stripped: with the ciphertext, or afterwards from the item left behind. +// Any other difference is rejected instead of becoming the next retry's source. +func projectReasoningRecoveryEdits(clean, before, after []byte) ([]byte, error) { if bytes.Equal(before, after) { return clean, nil } @@ -234,19 +235,59 @@ func projectReasoningCipherEdits(clean, before, after []byte) ([]byte, error) { if len(wireItems) != len(cleanItems) { return nil, errReasoningWireProjection } - wireIndices, cleanIndices := []int{}, []int{} + lost := func(index int, field string) bool { + return !gjson.GetBytes(after, fmt.Sprintf("input.%d.%s", index, field)).Exists() + } + var wireEdit, cleanEdit openAIReasoningRecoveryEdit for index, item := range wireItems { if item.hash != cleanItems[index].hash { return nil, errReasoningWireProjection } - if !gjson.GetBytes(after, fmt.Sprintf("input.%d.encrypted_content", item.index)).Exists() { - wireIndices = append(wireIndices, item.index) - cleanIndices = append(cleanIndices, cleanItems[index].index) + if !lost(item.index, "encrypted_content") { + continue + } + wireEdit.cipher = append(wireEdit.cipher, item.index) + cleanEdit.cipher = append(cleanEdit.cipher, cleanItems[index].index) + if item.id != "" && lost(item.index, "id") { + wireEdit.ids = append(wireEdit.ids, item.index) + cleanEdit.ids = append(cleanEdit.ids, cleanItems[index].index) + } + } + // An item left behind without ciphertext has no hash to pair it by. It is + // the same item in both bodies when it is the same occurrence of its id. + wireBare, cleanBare := openAIReasoningBareItemsByID(before), openAIReasoningBareItemsByID(clean) + for id, indices := range wireBare { + for occurrence, index := range indices { + if !lost(index, "id") { + continue + } + if occurrence >= len(cleanBare[id]) { + return nil, errReasoningWireProjection + } + wireEdit.ids = append(wireEdit.ids, index) + cleanEdit.ids = append(cleanEdit.ids, cleanBare[id][occurrence]) } } - expected, err := stripOpenAIReasoningCipherIndices(before, wireIndices) + expected, err := applyOpenAIReasoningRecoveryEdit(before, wireEdit) if err != nil || !bytes.Equal(expected, after) { return nil, errReasoningWireProjection } - return stripOpenAIReasoningCipherIndices(clean, cleanIndices) + return applyOpenAIReasoningRecoveryEdit(clean, cleanEdit) +} + +// openAIReasoningBareItemsByID lists, for each id, the input positions of the +// reasoning items that carry it and have no ciphertext. +func openAIReasoningBareItemsByID(body []byte) map[string][]int { + out := map[string][]int{} + input := gjson.GetBytes(body, "input") + if !input.IsArray() { + return out + } + for index, item := range input.Array() { + id := item.Get("id") + if item.Get("type").String() == "reasoning" && !openAIReasoningItemHasCipher(item) && id.Type == gjson.String && id.String() != "" { + out[id.String()] = append(out[id.String()], index) + } + } + return out } From bbe13955228e544c9d5e5f286a8ec7de4650d634 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 7 Oct 2026 12:21:30 +0800 Subject: [PATCH 2/3] fix(reasoning): edit stripped items in one pass and keep the stripped retry's record Follow-up to the id repair, from review. - Deleting a field from the stripped reasoning items edited the whole body once per item, in the strip, in the repair and again in the projection. At the sizes seen in production (hundreds of items, megabytes) that is hundreds of milliseconds to seconds of CPU per request. The items are now edited on their own and the body is written once; the per-item lookups read the input array once. The bytes are those of the one-by-one deletion, which is kept as the fallback and as the oracle of the new equivalence test. - The repair no longer resets the record of the stripped retry (retryDispatched, diagnosticState): the continuation diagnostic goes on describing the stripped retry, as the handbook says. - The repaired answer's status and request id are recorded with its own event and are not left in the state, which matters for the Chat Completions path that does not observe each response. - Handbook: the bound of three POSTs is per forwarding attempt. Co-Authored-By: Claude Opus 5.5 --- .downstream/gpt6-models-and-prompts.md | 2 +- .../service/openai_reasoning_recovery.go | 80 ++++++++++++++++++- .../openai_reasoning_recovery_item_id_test.go | 80 ++++++++++++++++++- .../internal/service/system_prompt_openai.go | 6 +- 4 files changed, 160 insertions(+), 8 deletions(-) diff --git a/.downstream/gpt6-models-and-prompts.md b/.downstream/gpt6-models-and-prompts.md index bdeadfd750ca..f78ee737cf0f 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 预算。去掉密文后留下的推理项仍带着 `id`，官方上游可能把它当作对已存储条目的引用去查，查不到时返回 `Item with id '' not found`（官方为 HTTP 404；生产上只见于 `rs_` 加 32 位十六进制的 id，推断由中转签发，OpenAI 自己签发的 id 留着被接受）。上游的错误消息这样点名、且被点名的正是本次尝试去掉过密文的推理项时，把本次尝试去掉过密文的推理项的 `id` 一并去掉（条目和其余字段保留；没有 `summary` 的条目去掉 `id` 后会被上游拒收，所以它保留 `id`，被点名的是它时不修），在同一账号再发一次。这一次与上面的去密文重发各算各的、各自最多一次，所以同一账号最多发三次（首次、去密文、去 id）；先去 id 再遇到密文被拒时，仍可去密文重发一次，届时被去密文的条目同时去掉 `id`。状态为 401/402/403/407/429 时不修；其他 4xx/5xx 只要消息点名了这样的条目就修（中转可能改写状态码）。流内失败不触发。这一步不写任何记忆：同一会话之后的每个请求仍先按原样发出、被拒后再修一次，即每个受影响的请求多一次很快被拒的上传。被修掉的那个错误应答不影响账号健康；修正后的请求若仍失败，按它所处的位置处理（未做过去密文重发时按首次发送，做过时按去密文重发的规则）。原生 Responses、自动透传以及 Chat Completions 端点上的 Responses 形状请求都适用。Ops 记为 `retry_without_item_ids`，续接诊断的 `recovery.item_ids_removed` 是被去掉 `id` 的条目数，`disposition` 仍只描述去密文的那一次重发。官方 API 是否接受不带 `id` 的无密文推理项未用真实上游验证过（Codex OAuth 后端已验证接受）；不接受时修正后的请求以新的错误结束，结果不比修之前差。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出因密文读不了而交给下一个账号的尝试。恢复重发因账号自身的原因失败时也换号（同样只在 `POST /v1/responses`，含 compact 与自动透传；Chat Completions 转 Responses 不换号）：重发已发出、失败属于该路径首次发送时就会换号的那一类，且客户端尚未收到本次尝试的任何输出时，请求按普通账号失败交给下一个账号——下一个账号收到原样请求，本账号不再重试，换号预算同普通换号。哪些失败算这一类以各路径首次发送的换号规则为准：非透传包括额度用尽、限流、鉴权失败、5xx 与流内的上游失败；自动透传的规则更窄，OAuth / Setup Token 透传的普通 5xx、不带账号状态信息的 401/403 等不在其内，仍在本账号结束。Ops 续接诊断用 `recovery.account_failure` 标出这样交给下一个账号的尝试。账号健康：失败的分类与首次发送相同（请求预算被拒 402/403 `balance_insufficient` 只算本次请求，结构化的 `model_not_supported` 只算该账号与模型，`budget_exceeded` 429 算账号，其余池模式可重试的状态不计入 API Key 健康熔断），调度失败照常上报，流内失败照常写入账号状态；重发的 HTTP 错误应答则不执行转发层在首次发送时的账号状态写入（标记限流、置为错误、停用、预算用尽标记、API Key 的瞬时失败计数），也不开启 OAuth 429 的同账号重试窗口，留给之后的首次发送处理，其中首次发送因同时停用账号而把池模式可重试状态计入账号的情形按不计入处理。处理器侧同账号重试用尽后的冷却照常调用，它只对 OAuth / Setup Token 账号起作用（硬额度 429 会写入额度状态）；OpenAI API Key 账号在 HTTP 错误应答下不被冷却，下一个请求仍可能先发给它，由那次首次发送写入状态。重发遇到请求校验错误、拒绝类结果、传输错误、响应头超时或未归类为可换号的错误时，仍在本账号结束；重发再次被报告密文无法解密时，按上文的密文规则换号。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 预算。去掉密文后留下的推理项仍带着 `id`，官方上游可能把它当作对已存储条目的引用去查，查不到时返回 `Item with id '' not found`（官方为 HTTP 404；生产上只见于 `rs_` 加 32 位十六进制的 id，推断由中转签发，OpenAI 自己签发的 id 留着被接受）。上游的错误消息这样点名、且被点名的正是本次尝试去掉过密文的推理项时，把本次尝试去掉过密文的推理项的 `id` 一并去掉（条目和其余字段保留；没有 `summary` 的条目去掉 `id` 后会被上游拒收，所以它保留 `id`，被点名的是它时不修），在同一账号再发一次。这一次与上面的去密文重发各算各的、各自最多一次，所以一次转发尝试里同一账号最多发三次（首次、去密文、去 id）；处理器对同一账号做同账号重试时，新的一次尝试从原样请求重新开始，这三次也重新计；先去 id 再遇到密文被拒时，仍可去密文重发一次，届时被去密文的条目同时去掉 `id`。状态为 401/402/403/407/429 时不修；其他 4xx/5xx 只要消息点名了这样的条目就修（中转可能改写状态码）。流内失败不触发。这一步不写任何记忆：同一会话之后的每次转发尝试仍先按原样发出、被拒后再修一次，即每次受影响的尝试多一次很快被拒的上传。被修掉的那个错误应答不影响账号健康；修正后的请求若仍失败，按它所处的位置处理（未做过去密文重发时按首次发送，做过时按去密文重发的规则）。原生 Responses、自动透传以及 Chat Completions 端点上的 Responses 形状请求都适用。Ops 记为 `retry_without_item_ids`，续接诊断的 `recovery.item_ids_removed` 是被去掉 `id` 的条目数，`disposition` 仍只描述去密文的那一次重发。官方 API 是否接受不带 `id` 的无密文推理项未用真实上游验证过（Codex OAuth 后端已验证接受）；不接受时修正后的请求以新的错误结束，结果不比修之前差。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出因密文读不了而交给下一个账号的尝试。恢复重发因账号自身的原因失败时也换号（同样只在 `POST /v1/responses`，含 compact 与自动透传；Chat Completions 转 Responses 不换号）：重发已发出、失败属于该路径首次发送时就会换号的那一类，且客户端尚未收到本次尝试的任何输出时，请求按普通账号失败交给下一个账号——下一个账号收到原样请求，本账号不再重试，换号预算同普通换号。哪些失败算这一类以各路径首次发送的换号规则为准：非透传包括额度用尽、限流、鉴权失败、5xx 与流内的上游失败；自动透传的规则更窄，OAuth / Setup Token 透传的普通 5xx、不带账号状态信息的 401/403 等不在其内，仍在本账号结束。Ops 续接诊断用 `recovery.account_failure` 标出这样交给下一个账号的尝试。账号健康：失败的分类与首次发送相同（请求预算被拒 402/403 `balance_insufficient` 只算本次请求，结构化的 `model_not_supported` 只算该账号与模型，`budget_exceeded` 429 算账号，其余池模式可重试的状态不计入 API Key 健康熔断），调度失败照常上报，流内失败照常写入账号状态；重发的 HTTP 错误应答则不执行转发层在首次发送时的账号状态写入（标记限流、置为错误、停用、预算用尽标记、API Key 的瞬时失败计数），也不开启 OAuth 429 的同账号重试窗口，留给之后的首次发送处理，其中首次发送因同时停用账号而把池模式可重试状态计入账号的情形按不计入处理。处理器侧同账号重试用尽后的冷却照常调用，它只对 OAuth / Setup Token 账号起作用（硬额度 429 会写入额度状态）；OpenAI API Key 账号在 HTTP 错误应答下不被冷却，下一个请求仍可能先发给它，由那次首次发送写入状态。重发遇到请求校验错误、拒绝类结果、传输错误、响应头超时或未归类为可换号的错误时，仍在本账号结束；重发再次被报告密文无法解密时，按上文的密文规则换号。Messages 桥接与 WebSocket 不恢复。关闭时不重发也不预先去除，签名拒绝仍作为请求级终态返回，不切换账号。 迁移 `267_purge_account_reasoning_options.sql` 一次性删除各账号 `extra` 中原有的两个账号级推理选项键（键名见该文件）；此后没有代码读写它们，已存的账号级取值不带入全局开关。Chat Completions 转 Responses 的请求只由客户端消息转换而来，网关不保存响应中的推理项。 diff --git a/backend/internal/service/openai_reasoning_recovery.go b/backend/internal/service/openai_reasoning_recovery.go index 05919b2677ed..85353416556f 100644 --- a/backend/internal/service/openai_reasoning_recovery.go +++ b/backend/internal/service/openai_reasoning_recovery.go @@ -397,9 +397,13 @@ func (r *openAIReasoningRecoveryState) skipRejectedHistory(body []byte) []byte { // stripped by this attempt. After the repair the id goes with the ciphertext. func (r *openAIReasoningRecoveryState) strip(body []byte, items []openAIReasoningCipherItem) ([]byte, error) { edit := openAIReasoningRecoveryEdit{} + var sent []gjson.Result + if r.dropItemIDs { + sent = gjson.GetBytes(body, "input").Array() + } for _, item := range items { edit.cipher = append(edit.cipher, item.index) - if r.dropItemIDs && item.id != "" && openAIReasoningItemStandsWithoutID(gjson.GetBytes(body, fmt.Sprintf("input.%d", item.index))) { + if r.dropItemIDs && item.id != "" && item.index < len(sent) && openAIReasoningItemStandsWithoutID(sent[item.index]) { edit.ids = append(edit.ids, item.index) } } @@ -534,7 +538,64 @@ func stripOpenAIReasoningCipherIndices(body []byte, indices []int) ([]byte, erro return deleteOpenAIReasoningItemField(body, indices, "encrypted_content") } +// deleteOpenAIReasoningItemField deletes one field from the reasoning items at +// the listed input positions. A long history holds hundreds of them, and +// editing the whole body once for each costs seconds, so the items are edited +// on their own and the body is written once. The result is byte for byte that +// of deleting the fields one by one. func deleteOpenAIReasoningItemField(body []byte, indices []int, field string) ([]byte, error) { + input := gjson.GetBytes(body, "input") + if len(indices) == 0 || !input.IsArray() { + return deleteOpenAIReasoningItemFieldOneByOne(body, indices, field) + } + listed := make(map[int]bool, len(indices)) + for _, index := range indices { + if listed[index] { + // Listed twice, an item is edited twice; no caller does that. + return deleteOpenAIReasoningItemFieldOneByOne(body, indices, field) + } + listed[index] = true + } + out := make([]byte, 0, len(body)) + position, copied, met, located := 0, 0, 0, true + var failed error + input.ForEach(func(_, item gjson.Result) bool { + if listed[position] { + if item.Get("type").String() != "reasoning" { + failed = errors.New("reasoning recovery target mismatch") + return false + } + end := item.Index + len(item.Raw) + if item.Index < copied || end > len(body) || string(body[item.Index:end]) != item.Raw { + // The item is not where its offset says: edit the slow way. + located = false + return false + } + edited, err := sjson.Delete(item.Raw, field) + if err != nil { + failed = err + return false + } + out = append(out, body[copied:item.Index]...) + out = append(out, edited...) + copied = end + met++ + } + position++ + return true + }) + switch { + case failed != nil: + return nil, failed + case !located: + return deleteOpenAIReasoningItemFieldOneByOne(body, indices, field) + case met != len(listed): + return nil, errors.New("reasoning recovery target mismatch") + } + return append(out, body[copied:]...), nil +} + +func deleteOpenAIReasoningItemFieldOneByOne(body []byte, indices []int, field string) ([]byte, error) { out := bytes.Clone(body) for _, index := range indices { path := fmt.Sprintf("input.%d", index) @@ -682,11 +743,19 @@ func (r *openAIReasoningRecoveryState) TryRepairUnfoundItemIDs(status int, heade if !strings.Contains(message, "Item with id '") { return nil, false } + input := gjson.GetBytes(r.wire, "input") + if !input.IsArray() { + return nil, false + } + items := input.Array() found := false ids := make([]int, 0, len(r.stripped)) for _, item := range r.stripped { + if item.index >= len(items) { + continue + } // What was stripped is checked against what was sent. - sent := gjson.GetBytes(r.wire, fmt.Sprintf("input.%d", item.index)) + sent := items[item.index] id := sent.Get("id") if item.id == "" || sent.Get("type").String() != "reasoning" || openAIReasoningItemHasCipher(sent) || id.Type != gjson.String || id.String() != item.id || !openAIReasoningItemStandsWithoutID(sent) || @@ -707,14 +776,17 @@ func (r *openAIReasoningRecoveryState) TryRepairUnfoundItemIDs(status int, heade r.itemIDsRemoved += len(ids) if r.retryUsed { // The stripped retry was already sent; the send boundary accepts only - // the body it knows, and this one is not sent yet. + // the body it knows. r.retryBody = bytes.Clone(repaired) - r.retryDispatched, r.diagnosticState = false, "retry_prepared" } + // The answer is recorded with its own status and request id, and leaves + // neither behind for a caller that does not observe each response. + observedStatus, observedHeaders := r.responseStatus, r.responseHeaders if r.responseStatus == 0 { r.responseStatus, r.responseHeaders = status, headers.Clone() } r.record("retry_without_item_ids", status, "", payload, len(ids)) + r.responseStatus, r.responseHeaders = observedStatus, observedHeaders return repaired, true } diff --git a/backend/internal/service/openai_reasoning_recovery_item_id_test.go b/backend/internal/service/openai_reasoning_recovery_item_id_test.go index 91de6655aac8..f3caba8259d5 100644 --- a/backend/internal/service/openai_reasoning_recovery_item_id_test.go +++ b/backend/internal/service/openai_reasoning_recovery_item_id_test.go @@ -377,20 +377,91 @@ func TestOpenAIReasoningRecoveryRepairChecksWhatItStrippedAgainstWhatWasSent(t * require.False(t, repaired) }) } + t.Run("an_entry_that_no_longer_holds_does_not_block_the_others", func(t *testing.T) { + const second = "rs_fedcba9876543210fedcba9876543210" + body := strings.Replace(reasoningRecoveryRelayFixture, `"output":"[]"}`, + `"output":"[]"},{"type":"reasoning","id":"`+second+`","summary":[],"encrypted_content":"opaque-second"}`, 1) + state, _, _ := newReasoningRecoveryTestState(t, context.Background(), body) + _, retry := state.TryRecover(http.StatusBadRequest, nil, []byte(`{"error":{"code":"invalid_encrypted_content"}}`), false) + require.True(t, retry) + state.wire, _ = sjson.SetBytes(state.retryBody, "input.1.type", "message") + + repaired, ok := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, []byte(`{"error":{"message":"Item with id '`+second+`' not found."}}`)) + require.True(t, ok) + require.Equal(t, reasoningRecoveryRelayItemID, gjson.GetBytes(repaired, "input.1.id").String()) + require.False(t, gjson.GetBytes(repaired, "input.4.id").Exists()) + }) t.Run("sent_as_stripped", func(t *testing.T) { state := prepared(t) state.MarkAttemptDispatched() state.wire = bytes.Clone(state.retryBody) - body, repaired := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) + state.responseStatus, state.responseHeaders = 0, nil + body, repaired := state.TryRepairUnfoundItemIDs(http.StatusNotFound, http.Header{"X-Request-Id": []string{"req_unfound"}}, unfound) require.True(t, repaired) require.False(t, gjson.GetBytes(body, "input.1.id").Exists()) require.Equal(t, string(body), string(state.retryBody), "the send boundary accepts the repaired body") - require.False(t, state.retryDispatched, "which is not sent yet") + require.True(t, state.retryDispatched, "the stripped retry stays recorded as sent") + require.Equal(t, "retry_attempted", state.diagnosticState) + require.Zero(t, state.responseStatus, "the repaired answer is not left behind as the attempt's status") + require.Nil(t, state.responseHeaders) _, again := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, unfound) require.False(t, again) }) } +// The fields are deleted in one pass over the body. The bytes are those of +// deleting them one whole-body edit at a time, which is what a request was +// sent with before, and the same targets are refused. +func TestOpenAIReasoningRecoveryDeletesFieldsInOnePassAsOneByOne(t *testing.T) { + const ( + compact = `{"type":"reasoning","id":"rs_a","summary":[{"type":"summary_text","text":"id \"quoted\" \\ é"}],"encrypted_content":"a"}` + idFirst = `{"id":"rs_b","type":"reasoning","encrypted_content":"b","summary":[]}` + idLast = `{"type":"reasoning","summary":[],"encrypted_content":"c","id":"rs_c"}` + idOnly = `{"type":"reasoning","id":"rs_d"}` + bare = `{"type":"reasoning","summary":[{"id":"nested","encrypted_content":"nested"}]}` + spaced = "{\n \"type\" : \"reasoning\" ,\n \"id\" : \"rs_e\" ,\n \"encrypted_content\" : \"e\" ,\n \"summary\" : [ ]\n }" + message = `{"type":"message","id":"msg_1","role":"user","content":"hello"}` + ) + bodies := map[string]string{ + "compact": `{"model":"m","input":[` + strings.Join([]string{message, compact, idFirst, message, idLast, idOnly, bare, spaced}, ",") + `],"store":false}`, + "spaced": "{\n \"input\" : [\n " + strings.Join([]string{message, compact, idFirst, message, idLast, idOnly, bare, spaced}, " ,\n ") + "\n ] ,\n \"model\" : \"m\"\n}", + "object": `{"input":{"0":` + compact + `}}`, + "absent": `{"model":"m"}`, + } + targets := map[string][]int{ + "none": nil, "one": {1}, "all": {1, 2, 4, 5, 6, 7}, "unsorted_and_repeated": {7, 2, 2, 1, 7}, + "a_message": {1, 3}, "out_of_range": {1, 8}, "negative": {-1}, "first": {0}, + } + for bodyName, body := range bodies { + for targetName, indices := range targets { + for _, field := range []string{"encrypted_content", "id"} { + t.Run(strings.Join([]string{bodyName, targetName, field}, "/"), func(t *testing.T) { + want, wantErr := deleteOpenAIReasoningItemFieldOneByOne([]byte(body), indices, field) + got, gotErr := deleteOpenAIReasoningItemField([]byte(body), indices, field) + require.Equal(t, wantErr == nil, gotErr == nil, "one refuses what the other refuses") + require.Equal(t, string(want), string(got)) + }) + } + } + } + // A history of the size seen in production stays one pass. + items := make([]string, 0, 400) + for index := range 400 { + items = append(items, strings.Replace(compact, "rs_a", "rs_"+strings.Repeat("0", 28)+string(rune('a'+index%26))+string(rune('a'+index/26)), 1)) + } + long := []byte(`{"model":"m","input":[` + strings.Join(items, ",") + `]}`) + all := make([]int, 400) + for index := range all { + all[index] = index + } + want, err := deleteOpenAIReasoningItemFieldOneByOne(long, all, "id") + require.NoError(t, err) + got, err := deleteOpenAIReasoningItemField(long, all, "id") + require.NoError(t, err) + require.Equal(t, string(want), string(got)) + require.NotContains(t, string(got), `"id":"rs_`) +} + // A Responses-shaped body on the Chat Completions endpoint is stripped by the // same rules and is repaired the same way. func TestOpenAIReasoningRecoveryChatPathRepairsAStrippedItemTheUpstreamCannotFind(t *testing.T) { @@ -474,6 +545,11 @@ func TestOpenAIReasoningRecoveryProjectionRepeatsOnlyRecoveryEdits(t *testing.T) clean: body(message), before: body(bare("rs_a"), message), after: body(idless, message), }, + { + name: "an_item_removed", + clean: body(message, cipher("rs_a", "a")), before: body(message, cipher("rs_a", "a")), + after: body(message), + }, } { t.Run(test.name, func(t *testing.T) { projected, err := projectReasoningRecoveryEdits([]byte(test.clean), []byte(test.before), []byte(test.after)) diff --git a/backend/internal/service/system_prompt_openai.go b/backend/internal/service/system_prompt_openai.go index 54bc88f5a153..d6f909871404 100644 --- a/backend/internal/service/system_prompt_openai.go +++ b/backend/internal/service/system_prompt_openai.go @@ -235,8 +235,12 @@ func projectReasoningRecoveryEdits(clean, before, after []byte) ([]byte, error) if len(wireItems) != len(cleanItems) { return nil, errReasoningWireProjection } + var afterItems []gjson.Result + if input := gjson.GetBytes(after, "input"); input.IsArray() { + afterItems = input.Array() + } lost := func(index int, field string) bool { - return !gjson.GetBytes(after, fmt.Sprintf("input.%d.%s", index, field)).Exists() + return index >= len(afterItems) || !afterItems[index].Get(field).Exists() } var wireEdit, cleanEdit openAIReasoningRecoveryEdit for index, item := range wireItems { From 1e6fe6359a0301a4caa6cdc342172626dd377e41 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 7 Oct 2026 15:38:36 +0800 Subject: [PATCH 3/3] fix(reasoning): track unsent ID repairs and their own response diagnostics --- .downstream/gpt6-models-and-prompts.md | 2 +- .../service/openai_reasoning_recovery.go | 43 ++++--- .../openai_reasoning_recovery_item_id_test.go | 110 +++++++++++++++++- 3 files changed, 136 insertions(+), 19 deletions(-) diff --git a/.downstream/gpt6-models-and-prompts.md b/.downstream/gpt6-models-and-prompts.md index f78ee737cf0f..3f42135d2534 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 预算。去掉密文后留下的推理项仍带着 `id`，官方上游可能把它当作对已存储条目的引用去查，查不到时返回 `Item with id '' not found`（官方为 HTTP 404；生产上只见于 `rs_` 加 32 位十六进制的 id，推断由中转签发，OpenAI 自己签发的 id 留着被接受）。上游的错误消息这样点名、且被点名的正是本次尝试去掉过密文的推理项时，把本次尝试去掉过密文的推理项的 `id` 一并去掉（条目和其余字段保留；没有 `summary` 的条目去掉 `id` 后会被上游拒收，所以它保留 `id`，被点名的是它时不修），在同一账号再发一次。这一次与上面的去密文重发各算各的、各自最多一次，所以一次转发尝试里同一账号最多发三次（首次、去密文、去 id）；处理器对同一账号做同账号重试时，新的一次尝试从原样请求重新开始，这三次也重新计；先去 id 再遇到密文被拒时，仍可去密文重发一次，届时被去密文的条目同时去掉 `id`。状态为 401/402/403/407/429 时不修；其他 4xx/5xx 只要消息点名了这样的条目就修（中转可能改写状态码）。流内失败不触发。这一步不写任何记忆：同一会话之后的每次转发尝试仍先按原样发出、被拒后再修一次，即每次受影响的尝试多一次很快被拒的上传。被修掉的那个错误应答不影响账号健康；修正后的请求若仍失败，按它所处的位置处理（未做过去密文重发时按首次发送，做过时按去密文重发的规则）。原生 Responses、自动透传以及 Chat Completions 端点上的 Responses 形状请求都适用。Ops 记为 `retry_without_item_ids`，续接诊断的 `recovery.item_ids_removed` 是被去掉 `id` 的条目数，`disposition` 仍只描述去密文的那一次重发。官方 API 是否接受不带 `id` 的无密文推理项未用真实上游验证过（Codex OAuth 后端已验证接受）；不接受时修正后的请求以新的错误结束，结果不比修之前差。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出因密文读不了而交给下一个账号的尝试。恢复重发因账号自身的原因失败时也换号（同样只在 `POST /v1/responses`，含 compact 与自动透传；Chat Completions 转 Responses 不换号）：重发已发出、失败属于该路径首次发送时就会换号的那一类，且客户端尚未收到本次尝试的任何输出时，请求按普通账号失败交给下一个账号——下一个账号收到原样请求，本账号不再重试，换号预算同普通换号。哪些失败算这一类以各路径首次发送的换号规则为准：非透传包括额度用尽、限流、鉴权失败、5xx 与流内的上游失败；自动透传的规则更窄，OAuth / Setup Token 透传的普通 5xx、不带账号状态信息的 401/403 等不在其内，仍在本账号结束。Ops 续接诊断用 `recovery.account_failure` 标出这样交给下一个账号的尝试。账号健康：失败的分类与首次发送相同（请求预算被拒 402/403 `balance_insufficient` 只算本次请求，结构化的 `model_not_supported` 只算该账号与模型，`budget_exceeded` 429 算账号，其余池模式可重试的状态不计入 API Key 健康熔断），调度失败照常上报，流内失败照常写入账号状态；重发的 HTTP 错误应答则不执行转发层在首次发送时的账号状态写入（标记限流、置为错误、停用、预算用尽标记、API Key 的瞬时失败计数），也不开启 OAuth 429 的同账号重试窗口，留给之后的首次发送处理，其中首次发送因同时停用账号而把池模式可重试状态计入账号的情形按不计入处理。处理器侧同账号重试用尽后的冷却照常调用，它只对 OAuth / Setup Token 账号起作用（硬额度 429 会写入额度状态）；OpenAI API Key 账号在 HTTP 错误应答下不被冷却，下一个请求仍可能先发给它，由那次首次发送写入状态。重发遇到请求校验错误、拒绝类结果、传输错误、响应头超时或未归类为可换号的错误时，仍在本账号结束；重发再次被报告密文无法解密时，按上文的密文规则换号。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 预算。去掉密文后留下的推理项仍带着 `id`，官方上游可能把它当作对已存储条目的引用去查，查不到时返回 `Item with id '' not found`（官方为 HTTP 404；生产上只见于 `rs_` 加 32 位十六进制的 id，推断由中转签发，OpenAI 自己签发的 id 留着被接受）。上游的错误消息这样点名、且被点名的正是本次尝试去掉过密文的推理项时，把本次尝试去掉过密文的推理项的 `id` 一并去掉（条目和其余字段保留；没有 `summary` 的条目去掉 `id` 后会被上游拒收，所以它保留 `id`，被点名的是它时不修），在同一账号再发一次。这一次与上面的去密文重发各算各的、各自最多一次，所以一次转发尝试里推理恢复最多让同一账号多收两次（去密文、去 id）；转发层其他的同账号重发（被拒字段的兼容重试、compact 模型回退、Agent Identity 任务恢复）另计，且只发生在去密文重发之前；处理器对同一账号做同账号重试时，新的一次尝试从原样请求重新开始，这两次也重新计；先去 id 再遇到密文被拒时，仍可去密文重发一次，届时被去密文的条目同时去掉 `id`。状态为 401/402/403/407/429 时不修；其他 4xx/5xx 只要消息点名了这样的条目就修（中转可能改写状态码）。流内失败不触发。这一步不写任何记忆：同一会话之后的每次转发尝试仍先按原样发出、被拒后再修一次，即每次受影响的尝试多一次很快被拒的上传。被修掉的那个错误应答不影响账号健康；修正后的请求若仍失败，按它所处的位置处理（未做过去密文重发时按首次发送，做过时按去密文重发的规则）。原生 Responses、自动透传以及 Chat Completions 端点上的 Responses 形状请求都适用。Ops 记为 `retry_without_item_ids`，续接诊断的 `recovery.item_ids_removed` 是被去掉 `id` 的条目数，`disposition` 仍只描述去密文的那一次重发。官方 API 是否接受不带 `id` 的无密文推理项未用真实上游验证过（Codex OAuth 后端已验证接受）；不接受时修正后的请求以新的错误结束，结果不比修之前差。本账号修不好时换号：在 `POST /v1/responses`（含 compact 与自动透传）上，上游报告请求里的密文无法解密或校验（结构化错误码、同类错误消息，或 `could not be decrypted` 的措辞），而本账号无法修复（不能恢复，或去掉推理密文后的重发、发送前已去掉密文的请求仍得到这类拒绝），且客户端尚未收到本次尝试的任何输出时，请求原样交给分组里的下一个账号——密文只能由签发它的上游组织解开，别的账号可能直接接受。每个请求最多因此换号 3 次（同时计入正常换号预算），下一个账号各自按上述规则恢复一次；被换下的账号不计健康惩罚，也不做同账号重试。换号上限或可选账号用尽时，客户端收到第一个报告密文问题的账号的那个错误；途中某个账号给出自己的终态（例如读得懂密文但请求本身无效）或网关侧容量限制时，返回该结果。没有账号给出回答时，被这次换号挪到后续账号上的粘性绑定回到第一个报告密文问题的账号，调度器保留在别处的绑定不动；有账号接受或已开始输出时，会话留在该账号。普通的请求校验错误、带 `previous_response_id` 或 `conversation` 的请求、上游状态为 401/402/403/407/429 时不换号；先试过透传账号再换到非透传账号时，沿用既有规则整块去掉加密推理项。经原生 Responses 路径转发的其他平台账号同样适用；Chat Completions 转 Responses 不换号。Ops 续接诊断用 `recovery.account_mismatch` 标出因密文读不了而交给下一个账号的尝试。恢复重发因账号自身的原因失败时也换号（同样只在 `POST /v1/responses`，含 compact 与自动透传；Chat Completions 转 Responses 不换号）：重发已发出、失败属于该路径首次发送时就会换号的那一类，且客户端尚未收到本次尝试的任何输出时，请求按普通账号失败交给下一个账号——下一个账号收到原样请求，本账号不再重试，换号预算同普通换号。哪些失败算这一类以各路径首次发送的换号规则为准：非透传包括额度用尽、限流、鉴权失败、5xx 与流内的上游失败；自动透传的规则更窄，OAuth / Setup Token 透传的普通 5xx、不带账号状态信息的 401/403 等不在其内，仍在本账号结束。Ops 续接诊断用 `recovery.account_failure` 标出这样交给下一个账号的尝试。账号健康：失败的分类与首次发送相同（请求预算被拒 402/403 `balance_insufficient` 只算本次请求，结构化的 `model_not_supported` 只算该账号与模型，`budget_exceeded` 429 算账号，其余池模式可重试的状态不计入 API Key 健康熔断），调度失败照常上报，流内失败照常写入账号状态；重发的 HTTP 错误应答则不执行转发层在首次发送时的账号状态写入（标记限流、置为错误、停用、预算用尽标记、API Key 的瞬时失败计数），也不开启 OAuth 429 的同账号重试窗口，留给之后的首次发送处理，其中首次发送因同时停用账号而把池模式可重试状态计入账号的情形按不计入处理。处理器侧同账号重试用尽后的冷却照常调用，它只对 OAuth / Setup Token 账号起作用（硬额度 429 会写入额度状态）；OpenAI API Key 账号在 HTTP 错误应答下不被冷却，下一个请求仍可能先发给它，由那次首次发送写入状态。重发遇到请求校验错误、拒绝类结果、传输错误、响应头超时或未归类为可换号的错误时，仍在本账号结束；重发再次被报告密文无法解密时，按上文的密文规则换号。Messages 桥接与 WebSocket 不恢复。关闭时不重发也不预先去除，签名拒绝仍作为请求级终态返回，不切换账号。 迁移 `267_purge_account_reasoning_options.sql` 一次性删除各账号 `extra` 中原有的两个账号级推理选项键（键名见该文件）；此后没有代码读写它们，已存的账号级取值不带入全局开关。Chat Completions 转 Responses 的请求只由客户端消息转换而来，网关不保存响应中的推理项。 diff --git a/backend/internal/service/openai_reasoning_recovery.go b/backend/internal/service/openai_reasoning_recovery.go index 85353416556f..c74ca2d365b0 100644 --- a/backend/internal/service/openai_reasoning_recovery.go +++ b/backend/internal/service/openai_reasoning_recovery.go @@ -97,8 +97,10 @@ type openAIReasoningRecoveryState struct { // dropItemIDs makes a strip remove the item's id with its ciphertext. It // is set by the repair, for the rest of this attempt. dropItemIDs bool - // idRepairUsed records the one re-send without the ids of stripped items. + // idRepairUsed records the one re-send without the ids of stripped items; + // idRepairUnsent, that its body has not been sent yet. idRepairUsed bool + idRepairUnsent bool itemIDsRemoved int } @@ -175,7 +177,11 @@ func (r *openAIReasoningRecoveryState) ObserveResponse(resp *http.Response) { } func (r *openAIReasoningRecoveryState) MarkAttemptDispatched() { - if r != nil && r.retryUsed { + if r == nil { + return + } + r.idRepairUnsent = false + if r.retryUsed { r.retryDispatched = true r.diagnosticState = "retry_attempted" } @@ -541,8 +547,9 @@ func stripOpenAIReasoningCipherIndices(body []byte, indices []int) ([]byte, erro // deleteOpenAIReasoningItemField deletes one field from the reasoning items at // the listed input positions. A long history holds hundreds of them, and // editing the whole body once for each costs seconds, so the items are edited -// on their own and the body is written once. The result is byte for byte that -// of deleting the fields one by one. +// on their own and the body is written once. For what its callers pass, valid +// JSON with one input and positions of that input, the result is byte for byte +// that of deleting the fields one by one. func deleteOpenAIReasoningItemField(body []byte, indices []int, field string) ([]byte, error) { input := gjson.GetBytes(body, "input") if len(indices) == 0 || !input.IsArray() { @@ -772,7 +779,7 @@ func (r *openAIReasoningRecoveryState) TryRepairUnfoundItemIDs(status int, heade if err != nil { return nil, false } - r.idRepairUsed, r.dropItemIDs = true, true + r.idRepairUsed, r.idRepairUnsent, r.dropItemIDs = true, true, true r.itemIDsRemoved += len(ids) if r.retryUsed { // The stripped retry was already sent; the send boundary accepts only @@ -782,9 +789,7 @@ func (r *openAIReasoningRecoveryState) TryRepairUnfoundItemIDs(status int, heade // The answer is recorded with its own status and request id, and leaves // neither behind for a caller that does not observe each response. observedStatus, observedHeaders := r.responseStatus, r.responseHeaders - if r.responseStatus == 0 { - r.responseStatus, r.responseHeaders = status, headers.Clone() - } + r.responseStatus, r.responseHeaders = status, headers.Clone() r.record("retry_without_item_ids", status, "", payload, len(ids)) r.responseStatus, r.responseHeaders = observedStatus, observedHeaders return repaired, true @@ -890,6 +895,12 @@ func (r *openAIReasoningRecoveryState) StopError(err error) error { r.accountMismatch = r.anotherAccountMayAccept(failure, signatureRejected) r.accountFailure = !r.accountMismatch && r.accountFailedAfterRetry(returned) if !r.retryUsed && !signatureRejected && !cacheSkipRejected && !r.accountMismatch { + if r.idRepairUnsent && !r.stopRecorded { + // A cached pre-strip needs no stripped retry. Keep the builder's + // error contract, but record that its prepared id repair was not sent. + r.stopRecorded = true + r.record("recovery_not_attempted", 0, "", nil, 0) + } return err } if failure == nil { @@ -940,9 +951,10 @@ func (r *openAIReasoningRecoveryState) anotherAccountMayAccept(failure *Upstream 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. + if (r.retryUsed && !r.retryDispatched) || r.idRepairUnsent { + // The stripped retry, or the repair after it, was prepared but never + // sent. That is a fault in how this request was built, not evidence + // about the account. return false } switch { @@ -962,7 +974,7 @@ func (r *openAIReasoningRecoveryState) anotherAccountMayAccept(failure *Upstream // rejection, continuation state and a refusal are not the account's, and an // error the attempt did not classify as a failover ends the request as before. func (r *openAIReasoningRecoveryState) accountFailedAfterRetry(returned *UpstreamFailoverError) bool { - if returned == nil || !r.switchAccounts || !r.retryUsed || !r.retryDispatched || + if returned == nil || !r.switchAccounts || !r.retryUsed || !r.retryDispatched || r.idRepairUnsent || r.semanticCommitted || r.failureTerminalForwarded || r.ctx.Err() != nil { return false } @@ -1011,13 +1023,14 @@ func (r *openAIReasoningRecoveryState) continuationDiagnosticRecovery(payload [] if shape.Disposition == "" { shape.Disposition = "not_attempted" } - if shape.Disposition == "not_attempted" { - shape.NotAttemptedReason = r.recoveryNotAttemptedReason(payload) - } else if r.retryUsed && !r.retryDispatched && r.stopRecorded { + if r.stopRecorded && (r.idRepairUnsent || (r.retryUsed && !r.retryDispatched)) { + // The last re-send prepared, stripped retry or repair, was not sent. shape.NotAttemptedReason = r.diagnosticStopReason if shape.NotAttemptedReason == "" { shape.NotAttemptedReason = "recovery_not_dispatched" } + } else if shape.Disposition == "not_attempted" { + shape.NotAttemptedReason = r.recoveryNotAttemptedReason(payload) } return shape } diff --git a/backend/internal/service/openai_reasoning_recovery_item_id_test.go b/backend/internal/service/openai_reasoning_recovery_item_id_test.go index f3caba8259d5..2d77b931b657 100644 --- a/backend/internal/service/openai_reasoning_recovery_item_id_test.go +++ b/backend/internal/service/openai_reasoning_recovery_item_id_test.go @@ -5,6 +5,8 @@ package service import ( "bytes" "context" + "errors" + "io" "net/http" "net/http/httptest" "strings" @@ -409,9 +411,111 @@ func TestOpenAIReasoningRecoveryRepairChecksWhatItStrippedAgainstWhatWasSent(t * }) } -// The fields are deleted in one pass over the body. The bytes are those of -// deleting them one whole-body edit at a time, which is what a request was -// sent with before, and the same targets are refused. +// The repair's event carries the status and the request id of the answer it +// repairs, whatever an earlier answer left in the state, and a repaired body +// that is then not sent is reported with the reason. +func TestOpenAIReasoningRecoveryRepairIsRecordedWithItsOwnAnswer(t *testing.T) { + state, req, _ := newReasoningRecoveryTestState(t, context.Background(), reasoningRecoveryRelayFixture) + state.BindDiagnosticRequest([]byte(reasoningRecoveryRelayFixture), req) + stripped, retry := state.TryRecover(http.StatusBadRequest, http.Header{"X-Request-Id": []string{"req_cipher_rejected"}}, []byte(`{"error":{"code":"invalid_encrypted_content"}}`), false) + require.True(t, retry) + retried, _, err := state.PrepareRequest(req.Clone(context.Background()), stripped, "") + require.NoError(t, err) + state.MarkAttemptDispatched() + + unfound := []byte(`{"error":{"message":"Item with id '` + reasoningRecoveryRelayItemID + `' not found."}}`) + repaired, ok := state.TryRepairUnfoundItemIDs(http.StatusNotFound, http.Header{"X-Request-Id": []string{"req_item_not_found"}}, unfound) + require.True(t, ok) + events, _ := state.c.Get(OpsUpstreamErrorsKey) + recorded := events.([]*OpsUpstreamErrorEvent) + last := recorded[len(recorded)-1] + require.Equal(t, "retry_without_item_ids", last.Message) + require.Equal(t, http.StatusNotFound, last.UpstreamStatusCode) + require.Equal(t, "req_item_not_found", last.UpstreamRequestID) + require.True(t, last.ContinuationDiagnostic.Recovery.RetryAttempted, "the stripped retry was sent") + + // The send boundary refuses a body other than the repaired one. + _, _, err = state.PrepareRequest(retried.Clone(context.Background()), stripped, "") + var stopped *OpenAIReasoningRecoveryTerminalError + require.ErrorAs(t, err, &stopped) + blocked := lastReasoningRecoveryDiagnostic(t, state.c).ContinuationDiagnostic.Recovery + require.Equal(t, "source_changed", blocked.NotAttemptedReason, "the repaired body was never sent") + require.True(t, blocked.RetryAttempted, "the stripped retry stays recorded as sent") + require.NotEqual(t, string(stripped), string(repaired)) +} + +// A repair prepared after either a cached strip or a sent stripped retry may +// fail before sending. Report that fact without handing the old answer to +// another account. The cached path keeps the builder's ordinary error contract. +func TestOpenAIReasoningRecoveryUnsentRepairKeepsItsDiagnostic(t *testing.T) { + for _, test := range []struct { + name string + cached bool + prior func() error + }{ + {name: "cached_snapshot_failure", cached: true}, + {name: "stripped_retry_snapshot_failure"}, + {name: "stripped_retry_old_account_failure", prior: func() error { + return &UpstreamFailoverError{StatusCode: http.StatusBadGateway, ResponseBody: []byte(`{"error":{"code":"credit_balance_exhausted"}}`)} + }}, + {name: "stripped_retry_old_cipher_rejection", prior: func() error { + return NewOpenAIContinuationStateUnavailableError(http.StatusBadRequest, nil, []byte(`{"error":{"code":"invalid_encrypted_content"}}`)) + }}, + } { + t.Run(test.name, func(t *testing.T) { + state, req, _ := newReasoningRecoveryTestState(t, context.Background(), reasoningRecoveryRelayFixture) + var body []byte + if test.cached { + store := &reasoningRecoveryMemoryStore{} + learner, _, _ := newReasoningRecoveryTestState(t, context.Background(), reasoningRecoveryRelayFixture) + learner.store = store + _, retry := learner.TryRecover(http.StatusBadRequest, nil, []byte(`{"error":{"code":"invalid_encrypted_content"}}`), false) + require.True(t, retry) + state.store = store + body = []byte(reasoningRecoveryRelayFixture) + } else { + var retry bool + body, retry = state.TryRecover(http.StatusBadRequest, nil, []byte(`{"error":{"code":"invalid_encrypted_content"}}`), false) + require.True(t, retry) + } + retried, _, err := state.PrepareRequest(req.Clone(context.Background()), body, "") + require.NoError(t, err) + state.BindDiagnosticRequest([]byte(reasoningRecoveryRelayFixture), retried) + state.MarkAttemptDispatched() + repaired, ok := state.TryRepairUnfoundItemIDs(http.StatusNotFound, nil, []byte(`{"error":{"message":"Item with id '`+reasoningRecoveryRelayItemID+`' not found."}}`)) + require.True(t, ok) + + var attempt error + if test.prior != nil { + attempt = test.prior() + } else { + unsent := retried.Clone(context.Background()) + unsent.GetBody = func() (io.ReadCloser, error) { return nil, errors.New("snapshot unavailable") } + _, _, attempt = state.PrepareRequest(unsent, repaired, "") + require.Error(t, attempt) + } + stopped := state.StopError(attempt) + if test.cached { + require.Same(t, attempt, stopped) + } else { + var terminal *OpenAIReasoningRecoveryTerminalError + require.ErrorAs(t, stopped, &terminal) + require.False(t, terminal.Failure.ShouldRetryNextAccount()) + } + diagnostic := lastReasoningRecoveryDiagnostic(t, state.c).ContinuationDiagnostic.Recovery + require.Equal(t, "recovery_not_dispatched", diagnostic.NotAttemptedReason) + require.Equal(t, !test.cached, diagnostic.RetryAttempted) + require.Equal(t, 1, diagnostic.ItemIDsRemoved) + require.False(t, diagnostic.AccountMismatch) + require.False(t, diagnostic.AccountFailure) + }) + } +} + +// The fields are deleted in one pass over the body. For valid JSON with one +// input, which is all the callers pass, the bytes are those of deleting them +// one whole-body edit at a time, which is what a request was sent with before, +// and the same targets are refused. func TestOpenAIReasoningRecoveryDeletesFieldsInOnePassAsOneByOne(t *testing.T) { const ( compact = `{"type":"reasoning","id":"rs_a","summary":[{"type":"summary_text","text":"id \"quoted\" \\ é"}],"encrypted_content":"a"}`