From 901a77cfb1097cc9808ee2f6a041b044d3c278dd Mon Sep 17 00:00:00 2001 From: li Date: Fri, 14 Aug 2026 10:33:47 +0800 Subject: [PATCH 01/10] =?UTF-8?q?fix(apicompat):=20Anthropic=E2=86=92Chat?= =?UTF-8?q?=20=E6=A1=A5=E5=9B=9E=E4=BC=A0=E5=B7=A5=E5=85=B7=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E7=9A=84=20thinking=EF=BC=8C=E4=BF=AE=E5=A4=8D=20Deep?= =?UTF-8?q?Seek=20=E5=A4=9A=E8=BD=AE=20400?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /v1/messages 打到只支持 Chat Completions 的 OpenAI 兼容上游时， anthropicAssistantToChatMessages 把历史 assistant 消息里的 thinking 块整块丢弃。 DeepSeek 的 thinking mode 要求产生工具调用的 reasoning_content 随该 assistant 消息回传，缺失即 400 —— 表现为「单轮正常，一进多轮工具对话必现失败」。 这些 thinking 块本来就是同一个文件的出站方向 (chatMessageToAnthropicBlocks) 用上游 reasoning_content 生成的，客户端只是 原样回传：桥自己造出来的东西，回来时被自己丢掉。 兄弟路径 Responses→Chat (buildChatMessagesFromItems 的 pendingReasoning) 早已 处理同一约束。本次把同一策略补到漏掉的这条桥上，作用域严格对齐：reasoning 只挂 到带 tool_calls 的 assistant 消息，纯文本轮次维持现状。 Fixes #5528 --- .../chatcompletions_anthropic_bridge.go | 35 ++- .../chatcompletions_anthropic_bridge_test.go | 5 +- ...tions_anthropic_reasoning_passback_test.go | 229 ++++++++++++++++++ 3 files changed, 266 insertions(+), 3 deletions(-) create mode 100644 backend/internal/pkg/apicompat/chatcompletions_anthropic_reasoning_passback_test.go diff --git a/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge.go b/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge.go index 1a62f5620f..47d4601c24 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge.go +++ b/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge.go @@ -249,8 +249,8 @@ func anthropicUserToChatMessages(raw json.RawMessage) ([]ChatMessage, error) { // anthropicAssistantToChatMessages handles an Anthropic assistant message. // Text content → assistant message content; tool_use blocks → tool_calls on the -// same assistant message; thinking blocks are dropped (Chat Completions has no -// inbound thinking field, matching anthropicAssistantToResponses). +// same assistant message; thinking blocks → reasoning_content, but only on a +// message that carries tool calls (see anthropicThinkingToReasoningContent). func anthropicAssistantToChatMessages(raw json.RawMessage) ([]ChatMessage, error) { // Plain string → single assistant message. var s string @@ -289,9 +289,40 @@ func anthropicAssistantToChatMessages(raw json.RawMessage) ([]ChatMessage, error }) } + msg.ReasoningContent = anthropicThinkingToReasoningContent(blocks, len(msg.ToolCalls) > 0) + return []ChatMessage{msg}, nil } +// anthropicThinkingToReasoningContent folds thinking blocks back into the +// Chat Completions reasoning_content field. +// +// chatMessageToAnthropicBlocks emits the upstream's reasoning_content as a +// thinking block on the way out, so a multi-turn client echoes it back on the +// next request; dropping it here made the bridge lose exactly what it had just +// produced. DeepSeek's thinking mode requires the reasoning_content that +// produced a tool call to be replayed on that assistant message and answers +// 400 otherwise, which is why buildChatMessagesFromItems already carries +// pendingReasoning onto assistant tool-call messages in the Responses→Chat +// bridge. hasToolCalls keeps the scope identical to that sibling: reasoning +// rides along with tool calls only, never on a plain assistant text turn. +// +// redacted_thinking blocks and signature-only placeholders carry no plaintext +// and contribute nothing. Multiple blocks join with "\n", matching +// extractResponsesReasoningText. +func anthropicThinkingToReasoningContent(blocks []AnthropicContentBlock, hasToolCalls bool) string { + if !hasToolCalls { + return "" + } + var parts []string + for _, b := range blocks { + if b.Type == "thinking" && b.Thinking != "" { + parts = append(parts, b.Thinking) + } + } + return strings.Join(parts, "\n") +} + // anthropicToolsToChatTools maps Anthropic tool definitions to Chat Completions // function tools. Server-side tools (web_search_*) are dropped — they have no // Chat Completions equivalent. diff --git a/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge_test.go b/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge_test.go index 00726f5ef7..eb979f70e0 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge_test.go +++ b/backend/internal/pkg/apicompat/chatcompletions_anthropic_bridge_test.go @@ -143,8 +143,11 @@ func TestAnthropicToChatCompletionsRequest_ThinkingDropped(t *testing.T) { out, err := AnthropicToChatCompletionsRequest(req) require.NoError(t, err) require.Len(t, out.Messages, 1) - // Only text survives; thinking is dropped + // Only text survives. Thinking is dropped because this turn carries no tool + // calls — reasoning rides along with tool calls only, matching the + // Responses→Chat bridge (see anthropicThinkingToReasoningContent). require.Equal(t, `"answer"`, string(out.Messages[0].Content)) + require.Empty(t, out.Messages[0].ReasoningContent) } func TestAnthropicToChatCompletionsRequest_ToolChoiceAuto(t *testing.T) { diff --git a/backend/internal/pkg/apicompat/chatcompletions_anthropic_reasoning_passback_test.go b/backend/internal/pkg/apicompat/chatcompletions_anthropic_reasoning_passback_test.go new file mode 100644 index 0000000000..007a5face6 --- /dev/null +++ b/backend/internal/pkg/apicompat/chatcompletions_anthropic_reasoning_passback_test.go @@ -0,0 +1,229 @@ +package apicompat + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +// issue #5528：/v1/messages 客户端(Claude Code 等)打到只会 Chat Completions 的 +// OpenAI 兼容上游时，历史 assistant 消息里的 thinking 块被整块丢弃。DeepSeek 的 +// thinking mode 要求产生工具调用的 reasoning_content 随该 assistant 消息回传， +// 于是「单轮正常、一进多轮工具对话必现 400」。 + +func anthropicAssistantMsg(t *testing.T, blocks string) *AnthropicRequest { + t.Helper() + return &AnthropicRequest{ + Model: "deepseek-v4-flash", + MaxTokens: 256, + Messages: []AnthropicMessage{ + {Role: "user", Content: json.RawMessage(`"what's the weather?"`)}, + {Role: "assistant", Content: json.RawMessage(blocks)}, + {Role: "user", Content: json.RawMessage(`[{"type":"tool_result","tool_use_id":"toolu_1","content":"sunny"}]`)}, + }, + } +} + +const anthropicThinkingToolTurn = `[ + {"type":"thinking","thinking":"user wants weather, call the tool"}, + {"type":"text","text":"checking"}, + {"type":"tool_use","id":"toolu_1","name":"get_weather","input":{"city":"SF"}} +]` + +func TestAnthropicToChatCompletionsRequest_ThinkingBecomesReasoningContentOnToolTurn(t *testing.T) { + out, err := AnthropicToChatCompletionsRequest(anthropicAssistantMsg(t, anthropicThinkingToolTurn)) + require.NoError(t, err) + + var assistant *ChatMessage + for i := range out.Messages { + if out.Messages[i].Role == "assistant" { + assistant = &out.Messages[i] + break + } + } + require.NotNil(t, assistant, "assistant message must survive the bridge") + require.Equal(t, "user wants weather, call the tool", assistant.ReasoningContent, + "产生工具调用的 thinking 必须作为 reasoning_content 回传，否则 DeepSeek 400") + require.Len(t, assistant.ToolCalls, 1) + require.Equal(t, `"checking"`, string(assistant.Content), "text/tool_use 处理保持不变") +} + +// 上游线格式才是上游看到的东西：字段没序列化出去，等于没修。 +func TestAnthropicToChatCompletionsRequest_ReasoningContentSerializesOnWire(t *testing.T) { + out, err := AnthropicToChatCompletionsRequest(anthropicAssistantMsg(t, anthropicThinkingToolTurn)) + require.NoError(t, err) + + payload, err := json.Marshal(out) + require.NoError(t, err) + require.Contains(t, string(payload), `"reasoning_content":"user wants weather, call the tool"`) +} + +// 闭环不变式：thinking 块本来就是本桥出站时用上游 reasoning_content 生成的 +// (chatMessageToAnthropicBlocks)，客户端只是原样回传。出站造、入站丢 = 自己丢自己的东西。 +func TestAnthropicChatBridge_ReasoningSurvivesOutboundInboundRoundTrip(t *testing.T) { + upstream := ChatMessage{ + Role: "assistant", + ReasoningContent: "step 1: need the weather tool", + Content: json.RawMessage(`"checking"`), + ToolCalls: []ChatToolCall{{ + ID: "call_1", + Type: "function", + Function: ChatFunctionCall{Name: "get_weather", Arguments: `{"city":"SF"}`}, + }}, + } + + // 出站：Chat 响应 → Anthropic content blocks + blocks := chatMessageToAnthropicBlocks(upstream) + require.Equal(t, "thinking", blocks[0].Type) + require.Equal(t, upstream.ReasoningContent, blocks[0].Thinking) + + // 客户端下一轮把同一组 blocks 原样回传 + raw, err := json.Marshal(blocks) + require.NoError(t, err) + + // 入站：Anthropic content blocks → Chat 请求 + back, err := anthropicAssistantToChatMessages(raw) + require.NoError(t, err) + require.Len(t, back, 1) + require.Equal(t, upstream.ReasoningContent, back[0].ReasoningContent, + "出站生成的 thinking 必须能原样还原回 reasoning_content") + require.Len(t, back[0].ToolCalls, 1) +} + +// 兄弟不变式：Responses→Chat 桥(buildChatMessagesFromItems 的 pendingReasoning) +// 早就把 reasoning 挂到带 tool_calls 的 assistant 消息上了。等价历史下两条桥必须一致。 +func TestAnthropicChatBridge_MatchesResponsesChatBridgeReasoningPlacement(t *testing.T) { + responsesReq := &ResponsesRequest{ + Model: "deepseek-v4-flash", + Input: json.RawMessage(`[ + {"type":"message","role":"user","content":[{"type":"input_text","text":"what's the weather?"}]}, + {"type":"reasoning","summary":[{"type":"summary_text","text":"call the tool"}]}, + {"type":"function_call","call_id":"call_1","name":"get_weather","arguments":"{\"city\":\"SF\"}"}, + {"type":"function_call_output","call_id":"call_1","output":"sunny"} + ]`), + } + viaResponses, err := ResponsesToChatCompletionsRequest(responsesReq) + require.NoError(t, err) + + viaAnthropic, err := AnthropicToChatCompletionsRequest(anthropicAssistantMsg(t, `[ + {"type":"thinking","thinking":"call the tool"}, + {"type":"tool_use","id":"toolu_1","name":"get_weather","input":{"city":"SF"}} + ]`)) + require.NoError(t, err) + + reasoningOnToolCallMessage := func(msgs []ChatMessage) string { + for _, m := range msgs { + if m.Role == "assistant" && len(m.ToolCalls) > 0 { + return m.ReasoningContent + } + } + return "" + } + require.Equal(t, "call the tool", reasoningOnToolCallMessage(viaResponses.Messages), + "前置条件：兄弟桥本来就带 reasoning_content") + require.Equal(t, reasoningOnToolCallMessage(viaResponses.Messages), + reasoningOnToolCallMessage(viaAnthropic.Messages), + "两条桥对等价历史必须产出同样的 reasoning_content 位置") +} + +// 作用域守卫：不带工具调用的纯文本轮次维持现状(与兄弟桥一致 —— reasoning 只随 +// 工具调用回传)，避免把 reasoning_content 撒到不需要它的上游请求上。 +func TestAnthropicToChatCompletionsRequest_ThinkingWithoutToolCallsStaysDropped(t *testing.T) { + req := &AnthropicRequest{ + Model: "deepseek-v4-flash", + MaxTokens: 100, + Messages: []AnthropicMessage{ + {Role: "assistant", Content: json.RawMessage( + `[{"type":"thinking","thinking":"secret thoughts"},{"type":"text","text":"answer"}]`)}, + }, + } + + out, err := AnthropicToChatCompletionsRequest(req) + require.NoError(t, err) + require.Len(t, out.Messages, 1) + require.Empty(t, out.Messages[0].ReasoningContent) + require.Equal(t, `"answer"`, string(out.Messages[0].Content)) + + payload, err := json.Marshal(out) + require.NoError(t, err) + require.NotContains(t, string(payload), "reasoning_content") +} + +func TestAnthropicThinkingToReasoningContent(t *testing.T) { + blocksOf := func(t *testing.T, raw string) []AnthropicContentBlock { + t.Helper() + var blocks []AnthropicContentBlock + require.NoError(t, json.Unmarshal([]byte(raw), &blocks)) + return blocks + } + + cases := []struct { + name string + raw string + hasToolCalls bool + want string + }{ + { + name: "single_thinking_block", + raw: `[{"type":"thinking","thinking":"a"}]`, + hasToolCalls: true, + want: "a", + }, + { + // 多个 thinking 块用 "\n" 连接，与 extractResponsesReasoningText 一致。 + name: "multiple_blocks_join_with_newline", + raw: `[{"type":"thinking","thinking":"a"},{"type":"text","text":"x"},{"type":"thinking","thinking":"b"}]`, + hasToolCalls: true, + want: "a\nb", + }, + { + // redacted_thinking 没有明文可回传。 + name: "redacted_thinking_has_no_plaintext", + raw: `[{"type":"redacted_thinking","signature":"abc"}]`, + hasToolCalls: true, + want: "", + }, + { + // 只带 signature 的 thinking 占位块(xAI/Codex 密文回放形态)同样无明文。 + name: "signature_only_thinking", + raw: `[{"type":"thinking","thinking":"","signature":"gAAAAxxx"}]`, + hasToolCalls: true, + want: "", + }, + { + name: "no_tool_calls_returns_empty", + raw: `[{"type":"thinking","thinking":"a"}]`, + hasToolCalls: false, + want: "", + }, + { + name: "no_thinking_blocks", + raw: `[{"type":"text","text":"x"}]`, + hasToolCalls: true, + want: "", + }, + { + name: "empty_blocks", + raw: `[]`, + hasToolCalls: true, + want: "", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, + anthropicThinkingToReasoningContent(blocksOf(t, tc.raw), tc.hasToolCalls)) + }) + } +} + +// 纯字符串形态的 assistant content 没有 blocks 可读，走早返回分支，不得 panic。 +func TestAnthropicAssistantToChatMessages_PlainStringContentUnaffected(t *testing.T) { + msgs, err := anthropicAssistantToChatMessages(json.RawMessage(`"just text"`)) + require.NoError(t, err) + require.Len(t, msgs, 1) + require.Empty(t, msgs[0].ReasoningContent) + require.Equal(t, `"just text"`, string(msgs[0].Content)) +} From 3c5553e253acfd01a147e0cca48cbf0cf36de38a Mon Sep 17 00:00:00 2001 From: li Date: Fri, 14 Aug 2026 11:05:29 +0800 Subject: [PATCH 02/10] =?UTF-8?q?fix(responses):=20=E7=BD=91=E5=85=B3?= =?UTF-8?q?=E5=90=88=E6=88=90=E7=9A=84=20Responses=20=E5=AF=B9=E8=B1=A1?= =?UTF-8?q?=E8=A1=A5=E9=BD=90=20created=5Fat=EF=BC=8C=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E4=B8=A5=E6=A0=BC=E5=AE=A2=E6=88=B7=E7=AB=AF=E5=8F=8D=E5=BA=8F?= =?UTF-8?q?=E5=88=97=E5=8C=96=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 严格的 Responses 客户端（Rust serde 系，如 Codex / Grok CLI）把 created_at 声明为 必填字段，缺失即 `missing field 'created_at'` 直接读不出响应。 apicompat.ResponsesResponse 此前根本没有这个字段，六个构造点全部不写： Chat→Responses 与 Anthropic→Responses 两座桥的非流式结果、response.created、 终止事件，以及两处手写的 response.failed 帧。 两个流式 state 早就在 New*State() 里采集了 Created 时间戳，只是没有出口—— 反方向（Responses→Chat）每个 chunk 都老老实实发 created，正方向直接丢在地上。 原生 Responses 透传走 gjson/sjson 字节级改写，保留上游原样，不受影响； 本次只补合成路径。 Fixes #5601 --- .../internal/handler/stream_error_event.go | 27 +-- .../handler/stream_error_event_test.go | 16 ++ .../anthropic_to_responses_response.go | 21 ++- .../chatcompletions_responses_bridge.go | 31 +++- .../responses_created_at_wire_test.go | 166 ++++++++++++++++++ backend/internal/pkg/apicompat/types.go | 17 +- .../service/openai_compact_stream_bridge.go | 8 +- ..._compact_stream_failure_created_at_test.go | 49 ++++++ 8 files changed, 299 insertions(+), 36 deletions(-) create mode 100644 backend/internal/pkg/apicompat/responses_created_at_wire_test.go create mode 100644 backend/internal/service/openai_compact_stream_failure_created_at_test.go diff --git a/backend/internal/handler/stream_error_event.go b/backend/internal/handler/stream_error_event.go index 2af098dbed..8ba6095cfb 100644 --- a/backend/internal/handler/stream_error_event.go +++ b/backend/internal/handler/stream_error_event.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "strings" + "time" "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey" "github.com/gin-gonic/gin" @@ -19,13 +20,16 @@ type responsesFailedError struct { // responsesFailedBody 对齐 apicompat.makeResponsesCompletedEvent 输出的 response 子对象字段集。 // Output 用空 slice（不是 nil）确保 marshal 为 `[]` 而非 `null`。 +// CreatedAt 不带 omitempty：严格客户端把它当必填字段，缺失会以 +// `missing field 'created_at'` 反序列化失败——那正是本文件要避免的"客户端读不懂终止事件"。 type responsesFailedBody struct { - ID string `json:"id"` - Object string `json:"object"` - Model string `json:"model,omitempty"` - Status string `json:"status"` - Output []any `json:"output"` - Error responsesFailedError `json:"error"` + ID string `json:"id"` + Object string `json:"object"` + CreatedAt int64 `json:"created_at"` + Model string `json:"model,omitempty"` + Status string `json:"status"` + Output []any `json:"output"` + Error responsesFailedError `json:"error"` } // responsesFailedEvent 是写入 SSE data 行的顶层结构。 @@ -61,11 +65,12 @@ func writeResponsesFailedSSE(c *gin.Context, errType, message string) bool { payload, err := json.Marshal(responsesFailedEvent{ Type: "response.failed", Response: responsesFailedBody{ - ID: synthesizeResponseID(c), - Object: "response", - Model: requestModel(c), - Status: "failed", - Output: []any{}, + ID: synthesizeResponseID(c), + Object: "response", + CreatedAt: time.Now().Unix(), + Model: requestModel(c), + Status: "failed", + Output: []any{}, Error: responsesFailedError{ Code: mapResponsesErrorCode(errType), Message: message, diff --git a/backend/internal/handler/stream_error_event_test.go b/backend/internal/handler/stream_error_event_test.go index f24cf97fff..187deda19d 100644 --- a/backend/internal/handler/stream_error_event_test.go +++ b/backend/internal/handler/stream_error_event_test.go @@ -227,6 +227,22 @@ func TestOpenAIHandleStreamingAwareError_BareResponsesRouteEmitsResponseFailed(t } // Synthesized response.failed id falls back to uuid when no request_id is present. +// issue #5601：严格的 Responses 客户端把 created_at 当必填字段，缺失即 +// `missing field 'created_at'`。合成的终止事件若解析不了，本文件存在的意义 +// （给客户端一个可识别的终止事件而不是盲重连）就落空了。 +func TestOpenAIHandleStreamingAwareError_ResponsesStreamingCarriesCreatedAt(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointResponses) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareError(c, http.StatusBadGateway, "upstream_error", "boom", true) + + resp, _ := parseResponsesFailedSSE(t, w.Body.String()) + raw, ok := resp["created_at"] + assert.True(t, ok, "response.failed 必须带 created_at") + createdAt, ok := raw.(float64) + assert.True(t, ok, "created_at 必须是数字，得到 %T", raw) + assert.Greater(t, int64(createdAt), int64(0), "created_at 必须是有效的 unix 时间戳") +} + func TestSynthesizeResponseID_FallbackUUID(t *testing.T) { c, _ := newGinContextForEndpoint(t, EndpointResponses) id := synthesizeResponseID(c) diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go index db0343cd73..d5a7057036 100644 --- a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go +++ b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go @@ -21,10 +21,13 @@ func AnthropicToResponsesResponse(resp *AnthropicResponse) *ResponsesResponse { id = generateResponsesID() } + // Anthropic responses carry no creation timestamp, so stamp now — the same + // synthesize-what-the-client-requires rule the generated id above follows. out := &ResponsesResponse{ - ID: id, - Object: "response", - Model: resp.Model, + ID: id, + Object: "response", + CreatedAt: time.Now().Unix(), + Model: resp.Model, } var outputs []ResponsesOutput @@ -551,11 +554,12 @@ func makeResponsesCreatedEvent(state *AnthropicEventToResponsesState) ResponsesS Type: "response.created", SequenceNumber: seq, Response: &ResponsesResponse{ - ID: state.ResponseID, - Object: "response", - Model: state.Model, - Status: "in_progress", - Output: []ResponsesOutput{}, + ID: state.ResponseID, + Object: "response", + CreatedAt: state.Created, + Model: state.Model, + Status: "in_progress", + Output: []ResponsesOutput{}, }, } } @@ -602,6 +606,7 @@ func makeResponsesCompletedEvent( Response: &ResponsesResponse{ ID: state.ResponseID, Object: "response", + CreatedAt: state.Created, Model: state.Model, Status: status, Output: outputs, diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go index 70dee035d8..c58caa1a56 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go @@ -1042,11 +1042,22 @@ func ChatCompletionsResponseToResponses(resp *ChatCompletionsResponse, model str id = generateResponsesID() } + // Carry the upstream's own creation timestamp when it sent one; otherwise + // stamp now, same fallback shape as the generated id above. + createdAt := int64(0) + if resp != nil { + createdAt = resp.Created + } + if createdAt <= 0 { + createdAt = time.Now().Unix() + } + out := &ResponsesResponse{ - ID: id, - Object: "response", - Model: model, - Status: "completed", + ID: id, + Object: "response", + CreatedAt: createdAt, + Model: model, + Status: "completed", } if resp == nil { out.Output = []ResponsesOutput{emptyResponsesMessageOutput()} @@ -1493,6 +1504,7 @@ func FinalizeChatCompletionsResponsesStream(state *ChatCompletionsToResponsesStr Response: &ResponsesResponse{ ID: state.ResponseID, Object: "response", + CreatedAt: state.Created, Model: state.Model, Status: status, Output: state.chatOutput(), @@ -1510,11 +1522,12 @@ func ensureChatToResponsesCreated(state *ChatCompletionsToResponsesStreamState) state.CreatedSent = true return []ResponsesStreamEvent{chatToResponsesEvent(state, "response.created", &ResponsesStreamEvent{ Response: &ResponsesResponse{ - ID: state.ResponseID, - Object: "response", - Model: state.Model, - Status: "in_progress", - Output: []ResponsesOutput{}, + ID: state.ResponseID, + Object: "response", + CreatedAt: state.Created, + Model: state.Model, + Status: "in_progress", + Output: []ResponsesOutput{}, }, })} } diff --git a/backend/internal/pkg/apicompat/responses_created_at_wire_test.go b/backend/internal/pkg/apicompat/responses_created_at_wire_test.go new file mode 100644 index 0000000000..48468409d6 --- /dev/null +++ b/backend/internal/pkg/apicompat/responses_created_at_wire_test.go @@ -0,0 +1,166 @@ +package apicompat + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +// issue #5601：严格的 Responses 客户端（Rust serde 系，如 Codex / Grok CLI）把 +// created_at 声明为必填字段，缺失即 `missing field 'created_at'` 反序列化失败。 +// 网关合成的 Responses 对象（Chat→Responses、Anthropic→Responses 两座桥）此前从不 +// 写这个字段——尽管两个流式 state 早就采集好了 Created 时间戳，只是没有出口。 +// 原生 Responses 透传走 gjson/sjson 字节级改写，不受影响。 + +// responseObjectOf 取出事件里的 response 子对象（按线格式，而不是按 Go 结构体）。 +func responseObjectOf(t *testing.T, evt ResponsesStreamEvent) map[string]any { + t.Helper() + m := marshalEvent(t, evt) + resp, ok := m["response"].(map[string]any) + require.True(t, ok, "event must carry a response object: %v", m) + return resp +} + +func requireCreatedAt(t *testing.T, resp map[string]any) int64 { + t.Helper() + raw, ok := resp["created_at"] + require.True(t, ok, "response 对象必须带 created_at，否则严格客户端直接反序列化失败") + value, ok := raw.(float64) + require.True(t, ok, "created_at 必须是数字，得到 %T", raw) + require.Greater(t, int64(value), int64(0), "created_at 必须是有效的 unix 时间戳") + return int64(value) +} + +// omitempty 陷阱守卫：created_at 为 0 时也必须出现在线格式里， +// 否则「字段存在」这件事就依赖于运行时恰好非零。 +func TestWire_CreatedAtPresentEvenAtZero(t *testing.T) { + resp := responseObjectOf(t, ResponsesStreamEvent{ + Type: "response.created", + Response: &ResponsesResponse{ID: "resp_1", Object: "response", Status: "in_progress"}, + }) + require.Contains(t, resp, "created_at", "created_at 不得带 omitempty") + require.EqualValues(t, 0, resp["created_at"]) +} + +// --------------------------------------------------------------------------- +// Chat Completions → Responses +// --------------------------------------------------------------------------- + +func TestChatCompletionsResponseToResponses_CarriesCreatedAt(t *testing.T) { + t.Run("uses_upstream_created_when_present", func(t *testing.T) { + out := ChatCompletionsResponseToResponses(&ChatCompletionsResponse{ + ID: "chatcmpl_1", + Created: 1700000000, + Model: "deepseek-v4-flash", + Choices: []ChatChoice{{Message: ChatMessage{Role: "assistant", Content: json.RawMessage(`"hi"`)}}}, + }, "deepseek-v4-flash", nil, false, nil) + require.EqualValues(t, 1700000000, out.CreatedAt, "上游给了 created 就照搬，不要另起时间") + }) + + t.Run("stamps_now_when_upstream_omits_created", func(t *testing.T) { + out := ChatCompletionsResponseToResponses(&ChatCompletionsResponse{ + ID: "chatcmpl_2", + Model: "deepseek-v4-flash", + Choices: []ChatChoice{{Message: ChatMessage{Role: "assistant", Content: json.RawMessage(`"hi"`)}}}, + }, "deepseek-v4-flash", nil, false, nil) + require.Greater(t, out.CreatedAt, int64(0)) + }) + + t.Run("nil_upstream_response_still_stamps", func(t *testing.T) { + out := ChatCompletionsResponseToResponses(nil, "deepseek-v4-flash", nil, false, nil) + require.Greater(t, out.CreatedAt, int64(0), "空上游响应也必须产出可解析的对象") + }) +} + +// 同一条流里 response.created 与终止事件必须报同一个 created_at +// （官方语义：created_at 是这次 response 的创建时刻，不随事件变化）。 +func TestChatCompletionsToResponsesStream_CreatedAtStableAcrossEvents(t *testing.T) { + state := NewChatCompletionsToResponsesStreamState("deepseek-v4-flash") + require.Greater(t, state.Created, int64(0), "前提：state 早就采集了时间戳") + + var chunk ChatCompletionsChunk + require.NoError(t, json.Unmarshal( + []byte(`{"choices":[{"index":0,"delta":{"content":"hi"}}]}`), &chunk)) + + events := ChatCompletionsChunkToResponsesEvents(&chunk, state) + events = append(events, FinalizeChatCompletionsResponsesStream(state)...) + + seen := map[string]int64{} + for _, evt := range events { + if evt.Response == nil { + continue + } + seen[evt.Type] = requireCreatedAt(t, responseObjectOf(t, evt)) + } + + require.Contains(t, seen, "response.created") + require.Contains(t, seen, "response.completed") + require.Equal(t, state.Created, seen["response.created"]) + require.Equal(t, seen["response.created"], seen["response.completed"], + "同一条流的 created_at 必须恒定") +} + +// --------------------------------------------------------------------------- +// Anthropic → Responses +// --------------------------------------------------------------------------- + +func TestAnthropicToResponsesResponse_StampsCreatedAt(t *testing.T) { + out := AnthropicToResponsesResponse(&AnthropicResponse{ + ID: "msg_1", + Type: "message", + Role: "assistant", + Model: "claude-sonnet-4-20250514", + Content: []AnthropicContentBlock{{Type: "text", Text: "hi"}}, + }) + require.Greater(t, out.CreatedAt, int64(0), + "Anthropic 响应不带时间戳，网关必须自己盖一个") +} + +func TestAnthropicEventToResponsesStream_CreatedAtStableAcrossEvents(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-20250514" + require.Greater(t, state.Created, int64(0), "前提：state 早就采集了时间戳") + + var events []ResponsesStreamEvent + for _, raw := range []string{ + `{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-sonnet-4-20250514","content":[],"usage":{"input_tokens":3,"output_tokens":0}}}`, + `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hi"}}`, + `{"type":"content_block_stop","index":0}`, + `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":2}}`, + `{"type":"message_stop"}`, + } { + var evt AnthropicStreamEvent + require.NoError(t, json.Unmarshal([]byte(raw), &evt)) + events = append(events, AnthropicEventToResponsesEvents(&evt, state)...) + } + events = append(events, FinalizeAnthropicResponsesStream(state)...) + + seen := map[string]int64{} + for _, evt := range events { + if evt.Response == nil { + continue + } + seen[evt.Type] = requireCreatedAt(t, responseObjectOf(t, evt)) + } + + require.Contains(t, seen, "response.created") + require.Contains(t, seen, "response.completed") + require.Equal(t, state.Created, seen["response.created"]) + require.Equal(t, seen["response.created"], seen["response.completed"], + "同一条流的 created_at 必须恒定") +} + +// ResponsesClientToolStreamRestorer 对部分事件走 unmarshal→re-marshal。 +// 结构体没有该字段时，上游带来的 created_at 会在这一步被静默抹掉。 +func TestResponsesStreamEvent_CreatedAtSurvivesUnmarshalRemarshal(t *testing.T) { + upstream := []byte(`{"type":"response.completed","response":{"id":"resp_9","object":"response",` + + `"created_at":1700000123,"model":"gpt-5.5","status":"completed","output":[]}}`) + + var evt ResponsesStreamEvent + require.NoError(t, json.Unmarshal(upstream, &evt)) + require.EqualValues(t, 1700000123, evt.Response.CreatedAt) + + require.EqualValues(t, 1700000123, requireCreatedAt(t, responseObjectOf(t, evt))) +} diff --git a/backend/internal/pkg/apicompat/types.go b/backend/internal/pkg/apicompat/types.go index bcc8cd481e..84175f2a89 100644 --- a/backend/internal/pkg/apicompat/types.go +++ b/backend/internal/pkg/apicompat/types.go @@ -339,12 +339,17 @@ func (t *ResponsesTool) UnmarshalJSON(data []byte) error { // ResponsesResponse is the non-streaming response from POST /v1/responses. type ResponsesResponse struct { - ID string `json:"id"` - Object string `json:"object"` // "response" - Model string `json:"model"` - Status string `json:"status"` // "completed" | "incomplete" | "failed" - Output []ResponsesOutput `json:"output"` - Usage *ResponsesUsage `json:"usage,omitempty"` + ID string `json:"id"` + Object string `json:"object"` // "response" + // CreatedAt is the unix creation timestamp. Strict Responses clients declare + // it non-optional and abort with `missing field 'created_at'` when it is + // absent, so it is always emitted — no omitempty. Same rule as ID (see the + // "clients treat it as required" fallback in ChatCompletionsResponseToAnthropic). + CreatedAt int64 `json:"created_at"` + Model string `json:"model"` + Status string `json:"status"` // "completed" | "incomplete" | "failed" + Output []ResponsesOutput `json:"output"` + Usage *ResponsesUsage `json:"usage,omitempty"` // incomplete_details is present when status="incomplete" IncompleteDetails *ResponsesIncompleteDetails `json:"incomplete_details,omitempty"` diff --git a/backend/internal/service/openai_compact_stream_bridge.go b/backend/internal/service/openai_compact_stream_bridge.go index bcac8304c2..1c36f9f27c 100644 --- a/backend/internal/service/openai_compact_stream_bridge.go +++ b/backend/internal/service/openai_compact_stream_bridge.go @@ -6,6 +6,7 @@ import ( "net/http" "strconv" "strings" + "time" "github.com/gin-gonic/gin" "github.com/google/uuid" @@ -117,8 +118,11 @@ func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, errType "response": map[string]any{ "id": "resp_" + strings.ReplaceAll(uuid.NewString(), "-", ""), "object": "response", - "status": "failed", - "output": []any{}, + // 严格客户端把 created_at 当必填字段，缺失会反序列化失败， + // 终止事件就白发了（退化成盲重连）。与 writeResponsesFailedSSE 对齐。 + "created_at": time.Now().Unix(), + "status": "failed", + "output": []any{}, "error": map[string]any{ "code": errType, "message": message, diff --git a/backend/internal/service/openai_compact_stream_failure_created_at_test.go b/backend/internal/service/openai_compact_stream_failure_created_at_test.go new file mode 100644 index 0000000000..a3b6f9bd78 --- /dev/null +++ b/backend/internal/service/openai_compact_stream_failure_created_at_test.go @@ -0,0 +1,49 @@ +//go:build unit + +package service + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// issue #5601：严格的 Responses 客户端把 created_at 当必填字段，缺失即 +// `missing field 'created_at'`。writeOpenAICompactSSEFailureMessage 存在的理由就是 +// 让 Codex 能把这帧识别成合法终止事件；解析不了就退化回它想避免的盲重连。 +func TestWriteOpenAICompactSSEFailureMessage_CarriesCreatedAt(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + + writeOpenAICompactSSEFailureMessage(c, http.StatusBadGateway, "upstream_error", "boom") + + body := rec.Body.String() + require.Contains(t, body, "event: response.failed") + + _, payload, found := strings.Cut(body, "data: ") + require.True(t, found, "SSE 帧必须带 data 行: %q", body) + + var event struct { + Type string `json:"type"` + Response struct { + ID string `json:"id"` + Object string `json:"object"` + CreatedAt int64 `json:"created_at"` + Status string `json:"status"` + } `json:"response"` + } + require.NoError(t, json.Unmarshal([]byte(strings.TrimSpace(payload)), &event)) + + require.Equal(t, "response.failed", event.Type) + require.Equal(t, "response", event.Response.Object) + require.Equal(t, "failed", event.Response.Status) + require.Greater(t, event.Response.CreatedAt, int64(0), + "response.failed 必须带有效的 created_at，否则严格客户端读不出这帧") +} From b7ec3cdadf0f935da3569d5f16aa81c7b3d12b75 Mon Sep 17 00:00:00 2001 From: William Drake Date: Sun, 16 Aug 2026 08:07:45 -0700 Subject: [PATCH 03/10] fix(openai): fail raw CC streams truncated before any terminal chunk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The raw Chat Completions passthrough returned a nil error whenever the upstream body ended, so an upstream that cut the SSE stream mid-generation surfaced as HTTP 200 with usage 0/0: the client kept a half-finished answer, the request was not counted as an SLA failure, and ops saw nothing. Observed against opencode.ai/zen/go (deepseek-v4-pro): 11 of 1346 streams in 24h recorded a first token but no final usage, alongside a concurrent `connection reset by peer` against the upstream edge. Track whether the upstream ever emitted a terminal signal — [DONE], a usage chunk, or a finish_reason — and treat a stream that ends without one as an upstream failure. Matching only [DONE] would misfire on compatible upstreams that EOF right after their last chunk and would drop billing for tokens that were actually delivered, so all three count. - nothing written to the client yet: UpstreamFailoverError, retried on another account with no client-visible half stream - bytes already on the wire: typed openAIUpstreamStreamReadError, so the handler appends an SSE error frame and marks the request an SLA failure Client cancel/disconnect keeps the previous semantics — there the upstream read error is indistinguishable from a truncation, so partial usage is still billed rather than blamed on the account. Co-Authored-By: Claude Opus 5 (1M context) --- .../openai_gateway_chat_completions_raw.go | 83 +++-- ...penai_gateway_chat_completions_raw_test.go | 320 ++++++++++++++++++ .../service/openai_raw_stream_truncation.go | 161 +++++++++ .../service/openai_stream_read_error.go | 12 + 4 files changed, 554 insertions(+), 22 deletions(-) create mode 100644 backend/internal/service/openai_raw_stream_truncation.go diff --git a/backend/internal/service/openai_gateway_chat_completions_raw.go b/backend/internal/service/openai_gateway_chat_completions_raw.go index 7427c6fc9c..00bbda0e51 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw.go +++ b/backend/internal/service/openai_gateway_chat_completions_raw.go @@ -285,6 +285,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions( clientOutputStarted := false pendingLines := make([]string, 0, 8) refusalDetector := newOpenAIChatSilentRefusalDetector(requestBodyLen) + var terminal openAIRawStreamTerminalState writeLine := func(line string) { if clientDisconnected { @@ -323,6 +324,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions( refusalDetector.ObserveSSELine(line) if payload, ok := extractOpenAISSEDataLine(line); ok { trimmedPayload := strings.TrimSpace(payload) + terminal.ObserveDataLine(trimmedPayload) if trimmedPayload != "[DONE]" { observer.ObserveOpenAI([]byte(payload), strings.TrimSpace(gjson.Get(payload, "type").String())) usageOnlyChunk := isOpenAIChatUsageOnlyStreamChunk(payload) @@ -350,14 +352,65 @@ func (s *OpenAIGatewayService) streamRawChatCompletions( } } - if err := scanner.Err(); err != nil { - if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - logger.L().Warn("openai chat_completions raw: stream read error", - zap.Error(err), - zap.String("request_id", requestID), - ) + resultWithUsage := func() *OpenAIForwardResult { + return &OpenAIForwardResult{ + RequestID: requestID, + Usage: usage, + Model: originalModel, + BillingModel: billingModel, + UpstreamModel: upstreamModel, + UpstreamResponseModel: observedUpstreamResponseModel(c), + UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), + UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), + ReasoningEffort: reasoningEffort, + ServiceTier: resolvedOpenAIUpstreamServiceTier(c, serviceTier), + Stream: true, + Duration: time.Since(startTime), + FirstTokenMs: firstTokenMs, } - } else if !clientDisconnected && !clientOutputStarted { + } + + scanErr := scanner.Err() + if scanErr != nil && !errors.Is(scanErr, context.Canceled) && !errors.Is(scanErr, context.DeadlineExceeded) { + logger.L().Warn("openai chat_completions raw: stream read error", + zap.Error(scanErr), + zap.String("request_id", requestID), + ) + } + + // 客户端取消/断开后上游读失败与上游截断不可区分（取消会连带取消上游请求）， + // 沿用既有语义：按已收到的用量正常收尾计费，不判为上游故障。 + clientAborted := clientDisconnected || + errors.Is(scanErr, context.Canceled) || + errors.Is(scanErr, context.DeadlineExceeded) + + // 上游在任何终止信号之前结束：连接被 reset（scanErr != nil）或干净 EOF。 + // 两者都不能再记成功——此前统一返回 nil error，把上游截断伪装成 + // `HTTP 200 + usage 0/0`，客户端收到半截回答且 Ops 侧完全无感。 + if !clientAborted && terminal.IsTruncated(clientOutputStarted) { + cause := scanErr + if cause == nil { + cause = ErrOpenAIUpstreamStreamTruncated + } + logger.L().Warn("openai chat_completions raw: upstream stream truncated before terminal chunk", + zap.Error(cause), + zap.String("request_id", requestID), + zap.Int64("account_id", account.ID), + zap.String("upstream_model", upstreamModel), + zap.Bool("saw_sse_data", terminal.sawDataLine), + zap.Bool("client_output_started", clientOutputStarted), + ) + if !clientOutputStarted { + // 响应头尚未提交：可以透明换号重试，客户端不会看到半截流。 + return nil, newOpenAIRawStreamTruncatedFailoverError(c, account, requestID, cause) + } + // 已写出语义字节：无法再 failover，改为带类型的上游错误。handler 会据此 + // 补发 SSE error 帧并把本次请求计入 SLA 失败。 + recordOpenAIRawStreamTruncation(c, account, requestID, cause, "http_error") + return resultWithUsage(), newOpenAIUpstreamStreamReadError(cause) + } + + if scanErr == nil && !clientDisconnected && !clientOutputStarted { if refusalDetector.IsSilentRefusal() { return nil, newOpenAISilentRefusalFailoverError(c, account, requestID) } @@ -380,21 +433,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions( } } - return &OpenAIForwardResult{ - RequestID: requestID, - Usage: usage, - Model: originalModel, - BillingModel: billingModel, - UpstreamModel: upstreamModel, - UpstreamResponseModel: observedUpstreamResponseModel(c), - UpstreamResponseModelConflict: observedUpstreamResponseModelConflict(c), - UpstreamResponseServiceTier: observedUpstreamResponseServiceTier(c), - ReasoningEffort: reasoningEffort, - ServiceTier: resolvedOpenAIUpstreamServiceTier(c, serviceTier), - Stream: true, - Duration: time.Since(startTime), - FirstTokenMs: firstTokenMs, - }, nil + return resultWithUsage(), nil } // ensureOpenAIChatStreamUsage 确保 raw Chat Completions 流式请求会让上游返回 usage。 diff --git a/backend/internal/service/openai_gateway_chat_completions_raw_test.go b/backend/internal/service/openai_gateway_chat_completions_raw_test.go index 8a83bbc7ea..2f0b8dcc08 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw_test.go +++ b/backend/internal/service/openai_gateway_chat_completions_raw_test.go @@ -687,6 +687,326 @@ func TestForwardAsRawChatCompletions_StripsEmptyToolCallIdentity(t *testing.T) { require.True(t, followUpSeen) } +// 上游在生成中途干净 EOF（无 [DONE]/usage/finish_reason）且已向客户端写出内容： +// 不能再记成 HTTP 200 成功，必须回带类型化的上游截断错误，由 handler 补 SSE error +// 帧并计入 SLA 失败。 +func TestForwardAsRawChatCompletions_TruncatedStreamAfterOutputFailsRequest(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_cut","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"half an ans"},"finish_reason":null}]}`, + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_truncated"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.Error(t, err) + require.NotNil(t, result, "已收字节的用量仍需带回，供 ops 记录首 token 时延") + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr), "已写出语义字节后不得再 failover") + + code, message, ok := OpenAIUpstreamStreamReadErrorDetails(err) + require.True(t, ok) + require.Equal(t, OpenAIUpstreamStreamTruncatedCode, code) + require.NotEmpty(t, message) + // 已写出的内容保持原样透传，客户端拿到的仍是它已经收到的那部分。 + require.Contains(t, rec.Body.String(), `"content":"half an ans"`) + require.NotContains(t, rec.Body.String(), "data: [DONE]") +} + +// 上游 200 但一个 SSE 字节都没发：响应头尚未提交，应换号重试而不是回 200 空流。 +func TestForwardAsRawChatCompletions_EmptyStreamBeforeOutputTriggersFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_empty"}}, + Body: io.NopCloser(strings.NewReader("")), + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.Nil(t, result) + var failoverErr *UpstreamFailoverError + require.True(t, errors.As(err, &failoverErr)) + require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) + require.Equal(t, OpenAIUpstreamStreamTruncatedCode, + gjson.GetBytes(failoverErr.ResponseBody, "error.code").String()) + require.True(t, failoverErr.ShouldRetryNextAccount()) + require.False(t, c.Writer.Written(), "换号重试前不得提交 200 响应头") + require.Empty(t, rec.Body.String()) +} + +// 传输层错误（Cloudflare edge reset 等）在写出后同样不能记成功，且分类要区别于 +// 干净 EOF，便于 ops 分辨 reset 与静默截断。 +func TestForwardAsRawChatCompletions_StreamReadErrorAfterOutputFailsRequest(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_reset"}}, + Body: &openAIChatStreamReadErrorCloser{ + payload: []byte(`data: {"id":"chatcmpl_reset","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"partial"}}]}` + "\n\n"), + err: errors.New("read tcp 172.18.0.4->172.65.90.23:443: read: connection reset by peer"), + }, + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.Error(t, err) + require.NotNil(t, result) + + code, _, ok := OpenAIUpstreamStreamReadErrorDetails(err) + require.True(t, ok) + require.Equal(t, OpenAIUpstreamStreamReadErrorCode, code) + require.Contains(t, rec.Body.String(), `"content":"partial"`) +} + +// 边界：缺 [DONE] 但收到了 usage 帧 —— 生成已完整，只是尾巴丢失。必须继续按成功 +// 计费，否则会误伤那些跑完就直接 EOF 的兼容上游并白送 token。 +func TestForwardAsRawChatCompletions_MissingDoneWithUsageStillSucceeds(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_nodone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}`, + "", + `data: {"id":"chatcmpl_nodone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[],"usage":{"prompt_tokens":11,"completion_tokens":6,"total_tokens":17}}`, + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_nodone_usage"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, 11, result.Usage.InputTokens) + require.Equal(t, 6, result.Usage.OutputTokens) +} + +// 边界：缺 [DONE] 与 usage，但末帧带 finish_reason —— 生成正常结束，同样不判截断。 +func TestForwardAsRawChatCompletions_MissingDoneWithFinishReasonStillSucceeds(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_finish","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"done"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_finish","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`, + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_nodone_finish"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.NoError(t, err) + require.NotNil(t, result) + require.Contains(t, rec.Body.String(), `"finish_reason":"stop"`) +} + +// openAIRawStreamDisconnectedWriter 模拟客户端已断开：raw 直转路径经 +// WriteString 写出，故两个方法都必须失败（只覆盖 Write 会被内嵌 writer 绕过）。 +type openAIRawStreamDisconnectedWriter struct { + gin.ResponseWriter +} + +func (w *openAIRawStreamDisconnectedWriter) Write([]byte) (int, error) { + return 0, errors.New("write failed: client disconnected") +} + +func (w *openAIRawStreamDisconnectedWriter) WriteString(string) (int, error) { + return 0, errors.New("write failed: client disconnected") +} + +// 客户端已断开时上游随后截断：两者不可区分，沿用既有语义按已收用量正常收尾计费， +// 不得把客户端离场记成上游故障。 +func TestForwardAsRawChatCompletions_ClientDisconnectTruncationStillBills(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Writer = &openAIRawStreamDisconnectedWriter{ResponseWriter: c.Writer} + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_gone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}`, + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_gone"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.NoError(t, err) + require.NotNil(t, result) +} + +// 客户端取消会连带取消上游请求，上游读因此报 context.Canceled：同样不判为上游截断。 +func TestForwardAsRawChatCompletions_ClientCancelTruncationStillBills(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_cancel"}}, + Body: &openAIChatStreamReadErrorCloser{ + payload: []byte(`data: {"id":"chatcmpl_cancel","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}` + "\n\n"), + err: context.Canceled, + }, + }} + + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") + require.NoError(t, err) + require.NotNil(t, result) +} + +func TestOpenAIRawStreamTerminalState(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + payloads []string + clientStarted bool + wantTerminated bool + wantTruncated bool + }{ + { + name: "done sentinel", + payloads: []string{`{"choices":[{"delta":{"content":"a"}}]}`, "[DONE]"}, + clientStarted: true, + wantTerminated: true, + }, + { + name: "usage chunk", + payloads: []string{`{"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":1}}`}, + clientStarted: true, + wantTerminated: true, + }, + { + name: "finish reason", + payloads: []string{`{"choices":[{"delta":{},"finish_reason":"length"}]}`}, + clientStarted: true, + wantTerminated: true, + }, + { + name: "null finish reason is not terminal", + payloads: []string{`{"choices":[{"delta":{"content":"a"},"finish_reason":null}]}`}, + clientStarted: true, + wantTruncated: true, + }, + { + name: "usage null is not terminal", + payloads: []string{`{"choices":[{"delta":{"content":"a"}}],"usage":null}`}, + clientStarted: true, + wantTruncated: true, + }, + { + // 上游对 stream 请求回了裸 JSON：无 data: 行，既有行为是原样透传。 + name: "non-sse body already forwarded", + payloads: nil, + clientStarted: true, + }, + { + name: "no bytes at all", + payloads: nil, + clientStarted: false, + wantTruncated: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + var state openAIRawStreamTerminalState + for _, payload := range tt.payloads { + state.ObserveDataLine(payload) + } + require.Equal(t, tt.wantTerminated, state.Terminated()) + require.Equal(t, tt.wantTruncated, state.IsTruncated(tt.clientStarted)) + }) + } +} + func TestForwardAsRawChatCompletions_ClientDisconnectDrainsUsage(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_raw_stream_truncation.go b/backend/internal/service/openai_raw_stream_truncation.go new file mode 100644 index 0000000000..678d5125b7 --- /dev/null +++ b/backend/internal/service/openai_raw_stream_truncation.go @@ -0,0 +1,161 @@ +package service + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/tidwall/gjson" +) + +// openAIRawStreamTruncatedUpstreamMessage 是 raw CC 直转路径上游截断的 ops 消息。 +const openAIRawStreamTruncatedUpstreamMessage = "Upstream Chat Completions stream ended before any terminal chunk" + +// openAIRawStreamTerminalState 记录 raw Chat Completions SSE 流是否收到过 +// **终止信号**。 +// +// 背景：CC 直转路径把上游 SSE 原样透传，此前只要 HTTP 状态是 200 就按成功收尾—— +// 上游中途断流（Cloudflare edge reset、后端 worker 掉线）会被伪装成 +// `HTTP 200 + usage 0/0`：客户端拿到半截回答，网关既不报错也不计入 SLA， +// Ops 侧完全不可见。 +// +// 三种终止信号任一出现即认为上游"讲完了"，只是尾巴可能丢失，不作截断处理： +// +// - [DONE] —— OpenAI CC 协议标准哨兵 +// - usage chunk —— include_usage 生效时的末尾用量帧（网关强制打开） +// - finish_reason —— 生成正常结束（stop/length/tool_calls/...） +// +// 只认 [DONE] 会误伤那些跑完最后一帧就直接 EOF 的兼容上游；只认 usage 会误伤 +// 不支持 include_usage 的上游。三者取并集，把误判压到"上游确实在生成中途被切断"。 +type openAIRawStreamTerminalState struct { + // sawDataLine 表示上游至少发过一行 `data:`，即响应确实是 SSE 语义流。 + // 非 SSE 响应体（上游对 stream 请求返回裸 JSON）不参与截断判定，保持既有透传行为。 + sawDataLine bool + sawDone bool + sawUsage bool + sawFinishReason bool +} + +// ObserveDataLine 从单行 SSE `data:` 载荷中提取终止信号。payload 需已 TrimSpace。 +func (t *openAIRawStreamTerminalState) ObserveDataLine(payload string) { + if t == nil { + return + } + t.sawDataLine = true + if payload == "[DONE]" { + t.sawDone = true + return + } + if usage := gjson.Get(payload, "usage"); usage.Exists() && usage.IsObject() { + t.sawUsage = true + } + if t.sawFinishReason { + return + } + for _, choice := range gjson.Get(payload, "choices").Array() { + // finish_reason 为 null 时 String() 返回空串，不算终止。 + if strings.TrimSpace(choice.Get("finish_reason").String()) != "" { + t.sawFinishReason = true + return + } + } +} + +// Terminated 表示上游给出过终止信号。 +func (t *openAIRawStreamTerminalState) Terminated() bool { + return t != nil && (t.sawDone || t.sawUsage || t.sawFinishReason) +} + +// IsTruncated 判定上游是否在任何终止信号之前结束。clientOutputStarted 用于放行 +// 非 SSE 响应体：那类响应本就没有 data: 行，既有行为是原样透传，不在本次判定范围内； +// 但"一个字节都没收到"的空 200 依然算截断。 +func (t *openAIRawStreamTerminalState) IsTruncated(clientOutputStarted bool) bool { + if t == nil || t.Terminated() { + return false + } + return t.sawDataLine || !clientOutputStarted +} + +// newOpenAIRawStreamTruncatedFailoverError 处理"上游截断且尚未向客户端写出任何 +// 字节"的情况：响应头还没提交，可以透明换号重试，客户端不会看到半截流。 +func newOpenAIRawStreamTruncatedFailoverError( + c *gin.Context, + account *Account, + upstreamRequestID string, + cause error, +) *UpstreamFailoverError { + recordOpenAIRawStreamTruncation(c, account, upstreamRequestID, cause, "failover") + + headers := http.Header{} + if id := strings.TrimSpace(upstreamRequestID); id != "" { + headers.Set("x-request-id", id) + } + return &UpstreamFailoverError{ + StatusCode: http.StatusBadGateway, + ResponseBody: openAIRawStreamTruncatedErrorBody(cause), + ResponseHeaders: headers, + } +} + +// recordOpenAIRawStreamTruncation 把上游截断记入 ops 上下文，使其在错误日志与 +// 账号健康度中可见——这正是此前"HTTP 200 假成功"丢掉的信息。 +func recordOpenAIRawStreamTruncation( + c *gin.Context, + account *Account, + upstreamRequestID string, + cause error, + kind string, +) { + if c == nil { + return + } + message := openAIRawStreamTruncatedMessage(cause) + platform := PlatformOpenAI + accountID := int64(0) + accountName := "" + if account != nil { + platform = account.Platform + accountID = account.ID + accountName = account.Name + } + + setOpsUpstreamError(c, http.StatusBadGateway, message, "") + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + Platform: platform, + AccountID: accountID, + AccountName: accountName, + UpstreamStatusCode: http.StatusBadGateway, + UpstreamRequestID: strings.TrimSpace(upstreamRequestID), + Kind: kind, + Message: message, + }) +} + +// openAIRawStreamTruncatedMessage 拼出 ops 消息：干净 EOF 没有底层错误可带， +// 传输层错误（connection reset / http2 stream error）则保留原因以便定位。 +func openAIRawStreamTruncatedMessage(cause error) string { + if cause == nil || errors.Is(cause, ErrOpenAIUpstreamStreamTruncated) { + return openAIRawStreamTruncatedUpstreamMessage + } + return openAIRawStreamTruncatedUpstreamMessage + ": " + cause.Error() +} + +// openAIRawStreamTruncatedErrorBody 构造 failover 错误体，code/message 与 +// 写出后走 openAIUpstreamStreamReadError 的客户端分类保持一致。 +func openAIRawStreamTruncatedErrorBody(cause error) []byte { + code, message := classifyOpenAIUpstreamStreamReadError(cause) + body, err := json.Marshal(map[string]any{ + "error": map[string]any{ + "type": "upstream_error", + "code": code, + "message": message, + }, + }) + if err != nil { + return []byte(`{"error":{"type":"upstream_error","code":"` + OpenAIUpstreamStreamTruncatedCode + + `","message":"Upstream response stream ended before completion"}}`) + } + return body +} diff --git a/backend/internal/service/openai_stream_read_error.go b/backend/internal/service/openai_stream_read_error.go index 9a008d8e05..9fda5708ee 100644 --- a/backend/internal/service/openai_stream_read_error.go +++ b/backend/internal/service/openai_stream_read_error.go @@ -12,8 +12,17 @@ const ( // when an upstream HTTP/2 response stream is reset after the request started. OpenAIUpstreamHTTP2StreamErrorCode = "upstream_http2_stream_error" OpenAIUpstreamStreamReadErrorCode = "upstream_stream_read_error" + // OpenAIUpstreamStreamTruncatedCode is returned when an upstream SSE stream + // closes *cleanly* before delivering any terminal signal. A clean EOF carries + // no transport error, so without this classification a truncated generation is + // indistinguishable from a successful one. + OpenAIUpstreamStreamTruncatedCode = "upstream_stream_truncated" ) +// ErrOpenAIUpstreamStreamTruncated marks an upstream SSE stream that ended at +// EOF — without a read error — before any terminal signal arrived. +var ErrOpenAIUpstreamStreamTruncated = errors.New("upstream stream ended before any terminal chunk") + type openAIUpstreamStreamReadError struct { cause error clientCode string @@ -61,6 +70,9 @@ func OpenAIUpstreamStreamReadErrorDetails(err error) (code, message string, ok b func classifyOpenAIUpstreamStreamReadError(err error) (code, message string) { if err != nil { + if errors.Is(err, ErrOpenAIUpstreamStreamTruncated) { + return OpenAIUpstreamStreamTruncatedCode, "Upstream response stream ended before completion" + } lower := strings.ToLower(err.Error()) // net/http's HTTP/2 stream error is unexported. Its stable text contains // "stream error: stream ID ..."; match only the transport signature and From 32064d39e70b3c705cdbf65b0aabaf38b0510809 Mon Sep 17 00:00:00 2001 From: hank Date: Tue, 25 Aug 2026 17:04:38 +0800 Subject: [PATCH 04/10] fix(openai): normalize cross-provider reasoning replay --- .../service/openai_gateway_forward.go | 11 +++ .../service/openai_gateway_request_body.go | 67 +++++++++++++++++++ ...nai_gateway_request_body_reasoning_test.go | 57 ++++++++++++++++ ...nai_responses_rejected_field_retry_test.go | 28 ++++++++ 4 files changed, 163 insertions(+) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 6487385000..ebf12675c9 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -173,6 +173,17 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body) } if account.IsOpenAI() && (account.IsOpenAIApiKey() || account.IsOpenAIOAuthLike()) { + normalizedReasoningBody, reasoningChanged, reasoningErr := normalizeOpenAIResponsesReasoningContentReplay(body) + if reasoningErr != nil { + return nil, fmt.Errorf("normalize OpenAI Responses reasoning content replay: %w", reasoningErr) + } + if reasoningChanged { + body = normalizedReasoningBody + originalBody = normalizedReasoningBody + requestView = newOpenAIRequestView(normalizedReasoningBody) + reqModel, reqStream, promptCacheKey = requestView.Model, requestView.Stream, requestView.PromptCacheKey + originalModel = reqModel + } sanitizedBody, changed, sanitizeErr := sanitizeOpenAIResponsesInputItemIDs(body) if sanitizeErr != nil { return nil, fmt.Errorf("sanitize OpenAI Responses input item IDs: %w", sanitizeErr) diff --git a/backend/internal/service/openai_gateway_request_body.go b/backend/internal/service/openai_gateway_request_body.go index 6272875cc5..23cc01d670 100644 --- a/backend/internal/service/openai_gateway_request_body.go +++ b/backend/internal/service/openai_gateway_request_body.go @@ -395,6 +395,67 @@ func openAIRequestBodyHasTools(body []byte) bool { return false } +// normalizeOpenAIResponsesReasoningContentReplay removes non-portable +// reasoning.content arrays before history is sent to a real OpenAI Responses +// endpoint. Compatible providers may return visible reasoning blocks there, +// while OpenAI accepts only an empty array when the item is replayed. +// +// Keep the reasoning item and its portable fields (summary, encrypted_content, +// ids, and opaque extensions). Callers scope this normalization to OpenAI +// destinations; compatible providers may still consume their own content. +func normalizeOpenAIResponsesReasoningContentReplay(body []byte) ([]byte, bool, error) { + input := gjson.GetBytes(body, "input") + if !input.IsArray() { + return body, false, nil + } + + needsNormalization := false + input.ForEach(func(_, item gjson.Result) bool { + if strings.TrimSpace(item.Get("type").String()) != "reasoning" { + return true + } + content := item.Get("content") + if content.IsArray() && len(content.Array()) > 0 { + needsNormalization = true + return false + } + return true + }) + if !needsNormalization { + return body, false, nil + } + + var reqBody map[string]any + if err := decodeOpenAIJSONUseNumber(body, &reqBody); err != nil { + return body, false, fmt.Errorf("normalize OpenAI reasoning content replay: %w", err) + } + items, ok := reqBody["input"].([]any) + if !ok { + return body, false, nil + } + changed := false + for _, rawItem := range items { + item, ok := rawItem.(map[string]any) + if !ok || strings.TrimSpace(firstNonEmptyString(item["type"])) != "reasoning" { + continue + } + content, ok := item["content"].([]any) + if !ok || len(content) == 0 { + continue + } + delete(item, "content") + changed = true + } + if !changed { + return body, false, nil + } + normalized, err := marshalOpenAIUpstreamJSON(reqBody) + if err != nil { + return body, false, fmt.Errorf("serialize normalized OpenAI reasoning content replay: %w", err) + } + return normalized, true, nil +} + func normalizeOpenAIAPIKeyStoreFalseReasoningReplay(body []byte, knownStoreFalse bool) ([]byte, bool, error) { if !knownStoreFalse && gjson.GetBytes(body, "store").Type != gjson.False { return body, false, nil @@ -955,6 +1016,12 @@ func normalizeOpenAIResponsesWebSocketCompatibilityBody(body []byte, account *Ac return body, false, err } } + if next, normalizedReasoningContent, err := normalizeOpenAIResponsesReasoningContentReplay(normalized); err != nil { + return body, false, err + } else if normalizedReasoningContent { + normalized = next + changed = true + } if account.IsOpenAIApiKey() { if next, normalizedParallel, err := normalizeOpenAIParallelToolCallsWithoutTools(normalized, responsesLite); err != nil { return body, false, err diff --git a/backend/internal/service/openai_gateway_request_body_reasoning_test.go b/backend/internal/service/openai_gateway_request_body_reasoning_test.go index 543d294880..cbe24815cd 100644 --- a/backend/internal/service/openai_gateway_request_body_reasoning_test.go +++ b/backend/internal/service/openai_gateway_request_body_reasoning_test.go @@ -293,3 +293,60 @@ func TestNormalizeOpenAIParallelToolCallsWithoutTools_KeepsResponsesLiteAddition require.False(t, changed) require.Equal(t, gjson.False, gjson.GetBytes(normalized, "parallel_tool_calls").Type) } + +func TestNormalizeOpenAIResponsesReasoningContentReplayStripsCrossProviderArray(t *testing.T) { + body := []byte(`{"model":"gpt-5.6-sol","input":[` + + `{"type":"message","role":"user","content":"one"},` + + `{"type":"message","role":"assistant","content":"two"},` + + `{"type":"function_call","call_id":"call_1","name":"lookup","arguments":"{}"},` + + `{"type":"function_call_output","call_id":"call_1","output":"ok"},` + + `{"type":"message","role":"user","content":"five"},` + + `{"type":"reasoning","id":"rs_provider","summary":[{"type":"summary_text","text":"portable"}],"content":[{"type":"reasoning_text","text":"visible reasoning"}],"opaque":9007199254740993},` + + `{"type":"message","role":"assistant","content":[{"type":"output_text","text":"answer"}]}` + + `]}`) + + normalized, changed, err := normalizeOpenAIResponsesReasoningContentReplay(body) + + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, "reasoning", gjson.GetBytes(normalized, "input.5.type").String()) + require.False(t, gjson.GetBytes(normalized, "input.5.content").Exists()) + require.Equal(t, "portable", gjson.GetBytes(normalized, "input.5.summary.0.text").String()) + require.Equal(t, "9007199254740993", gjson.GetBytes(normalized, "input.5.opaque").Raw) + require.Equal(t, "answer", gjson.GetBytes(normalized, "input.6.content.0.text").String()) +} + +func TestNormalizeOpenAIResponsesReasoningContentReplayKeepsPortableShapes(t *testing.T) { + for _, body := range []string{ + `{"input":[{"type":"reasoning","summary":[]}]}`, + `{"input":[{"type":"reasoning","content":[],"summary":[]}]}`, + `{"input":[{"type":"message","content":[{"type":"input_text","text":"keep"}]}]}`, + } { + normalized, changed, err := normalizeOpenAIResponsesReasoningContentReplay([]byte(body)) + require.NoError(t, err) + require.False(t, changed) + require.JSONEq(t, body, string(normalized)) + } +} + +func TestNormalizeOpenAIResponsesWebSocketCompatibilityBodyStripsReasoningContentOnlyForOpenAI(t *testing.T) { + body := []byte(`{"type":"response.create","model":"gpt-5.6-sol","store":true,"input":[{"type":"reasoning","summary":[{"type":"summary_text","text":"keep"}],"content":[{"type":"reasoning_text","text":"remove"}]}]}`) + for _, accountType := range []string{AccountTypeAPIKey, AccountTypeOAuth} { + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{ + Platform: PlatformOpenAI, + Type: accountType, + }, false) + require.NoError(t, err) + require.True(t, changed) + require.False(t, gjson.GetBytes(normalized, "input.0.content").Exists()) + require.Equal(t, "keep", gjson.GetBytes(normalized, "input.0.summary.0.text").String()) + } + + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{ + Platform: PlatformZhipu, + Type: AccountTypeAPIKey, + }, false) + require.NoError(t, err) + require.False(t, changed) + require.JSONEq(t, string(body), string(normalized)) +} diff --git a/backend/internal/service/openai_responses_rejected_field_retry_test.go b/backend/internal/service/openai_responses_rejected_field_retry_test.go index 9f794b1e57..4ed3e3ba22 100644 --- a/backend/internal/service/openai_responses_rejected_field_retry_test.go +++ b/backend/internal/service/openai_responses_rejected_field_retry_test.go @@ -538,6 +538,34 @@ func TestOpenAIGatewayService_APIKeyStripsAllIndexedNamespacesBeforeFirstForward require.False(t, gjson.GetBytes(upstream.bodies[0], "input.1.namespace").Exists()) } +func TestOpenAIGatewayServiceProactivelyStripsCrossProviderReasoningContent(t *testing.T) { + body := []byte(`{"model":"gpt-5.5","stream":false,"store":true,"input":[` + + `{"type":"message","role":"user","content":"one"},` + + `{"type":"message","role":"assistant","content":"two"},` + + `{"type":"function_call","call_id":"call_1","name":"lookup","arguments":"{}"},` + + `{"type":"function_call_output","call_id":"call_1","output":"ok"},` + + `{"type":"message","role":"user","content":"five"},` + + `{"type":"reasoning","summary":[{"type":"summary_text","text":"keep"}],"content":[{"type":"reasoning_text","text":"remove"}]}` + + `]}`) + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + newOpenAIRejectedFieldTestResponse(http.StatusOK, `{"output":[],"usage":{"input_tokens":1,"output_tokens":1,"input_tokens_details":{"cached_tokens":0}}}`), + }} + + result, err := newOpenAIRejectedFieldTestService(upstream).Forward( + context.Background(), + newOpenAIRejectedFieldTestContext(body), + newOpenAIRejectedFieldTestAccount(), + body, + ) + + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.bodies, 1, "reasoning content should be normalized before the first upstream request") + require.Equal(t, "reasoning", gjson.GetBytes(upstream.bodies[0], "input.5.type").String()) + require.False(t, gjson.GetBytes(upstream.bodies[0], "input.5.content").Exists()) + require.Equal(t, "keep", gjson.GetBytes(upstream.bodies[0], "input.5.summary.0.text").String()) +} + func TestOpenAIGatewayService_OpenAIHTTPStripsInputNamespacesBeforeFirstForward(t *testing.T) { accounts := []struct { name string From 50ba146298538b4b683ef38f88548985d6871f47 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Wed, 26 Aug 2026 15:04:17 +0800 Subject: [PATCH 05/10] fix(openai): preserve multimodal client tool outputs --- .../pkg/apicompat/responses_client_tools.go | 22 ++++++++ .../apicompat/responses_client_tools_test.go | 50 ++++++++++++++++++- ...nai_gateway_responses_client_tools_test.go | 28 +++++++++++ 3 files changed, 99 insertions(+), 1 deletion(-) diff --git a/backend/internal/pkg/apicompat/responses_client_tools.go b/backend/internal/pkg/apicompat/responses_client_tools.go index 5e831c1c52..c537761062 100644 --- a/backend/internal/pkg/apicompat/responses_client_tools.go +++ b/backend/internal/pkg/apicompat/responses_client_tools.go @@ -358,6 +358,9 @@ func normalizeClientToolOutput(item map[string]any) { if _, ok := output.(string); ok { return } + if isResponsesToolOutputContent(output) { + return + } if output == nil { item["output"] = "" return @@ -370,6 +373,25 @@ func normalizeClientToolOutput(item map[string]any) { item["output"] = string(encoded) } +func isResponsesToolOutputContent(output any) bool { + parts, ok := output.([]any) + if !ok || len(parts) == 0 { + return false + } + for _, part := range parts { + typed, ok := part.(map[string]any) + if !ok { + return false + } + switch stringValue(typed["type"]) { + case "input_text", "input_image", "input_file": + default: + return false + } + } + return true +} + // normalizeToolSearchOutput converts both tool_search output wire shapes into // the string output required by function_call_output. Older clients send an // output field directly; newer Codex clients return discovered definitions in diff --git a/backend/internal/pkg/apicompat/responses_client_tools_test.go b/backend/internal/pkg/apicompat/responses_client_tools_test.go index bfe3fd2eb7..dd8bb74fea 100644 --- a/backend/internal/pkg/apicompat/responses_client_tools_test.go +++ b/backend/internal/pkg/apicompat/responses_client_tools_test.go @@ -460,7 +460,55 @@ func TestAdaptResponsesClientToolsWithInheritedMapping_LowersFollowupHistoryWith output := requireResponsesClientToolValue[map[string]any](t, items[1]) require.Equal(t, "function_call_output", output["type"]) require.NotContains(t, output, "id") - require.JSONEq(t, `[{"text":"ok","type":"input_text"}]`, requireResponsesClientToolValue[string](t, output["output"])) + require.Equal(t, []any{map[string]any{"type": "input_text", "text": "ok"}}, output["output"]) +} + +func TestAdaptResponsesClientTools_NormalizesCustomToolOutput(t *testing.T) { + tests := []struct { + name string + output any + wantOutput any + }{ + { + name: "supported content parts remain an array", + output: []any{ + map[string]any{"type": "input_text", "text": "ok"}, + map[string]any{"type": "input_image", "image_url": "https://example.com/image.png"}, + map[string]any{"type": "input_file", "file_id": "file_123"}, + }, + wantOutput: []any{ + map[string]any{"type": "input_text", "text": "ok"}, + map[string]any{"type": "input_image", "image_url": "https://example.com/image.png"}, + map[string]any{"type": "input_file", "file_id": "file_123"}, + }, + }, + {name: "ordinary object is stringified", output: map[string]any{"ok": true}, wantOutput: `{"ok":true}`}, + {name: "arbitrary array is stringified", output: []any{"ok"}, wantOutput: `["ok"]`}, + {name: "empty array is stringified", output: []any{}, wantOutput: `[]`}, + {name: "mixed array is stringified", output: []any{map[string]any{"type": "input_text", "text": "ok"}, "bad"}, wantOutput: `[{"text":"ok","type":"input_text"},"bad"]`}, + {name: "unknown content type is stringified", output: []any{map[string]any{"type": "output_text", "text": "bad"}}, wantOutput: `[{"text":"bad","type":"output_text"}]`}, + {name: "whitespace-padded content type is stringified", output: []any{map[string]any{"type": " input_text ", "text": "bad"}}, wantOutput: `[{"text":"bad","type":" input_text "}]`}, + {name: "missing content type is stringified", output: []any{map[string]any{"text": "bad"}}, wantOutput: `[{"text":"bad"}]`}, + {name: "non-string content type is stringified", output: []any{map[string]any{"type": 1}}, wantOutput: `[{"type":1}]`}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + req := map[string]any{ + "tools": []any{map[string]any{"type": "custom", "name": "exec"}}, + "input": []any{map[string]any{ + "type": "custom_tool_call_output", "call_id": "call_1", "output": tc.output, + }}, + } + + _, changed, err := AdaptResponsesClientTools(req) + require.NoError(t, err) + require.True(t, changed) + item := requireResponsesClientToolValue[map[string]any](t, requireResponsesClientToolValue[[]any](t, req["input"])[0]) + require.Equal(t, "function_call_output", item["type"]) + require.Equal(t, tc.wantOutput, item["output"]) + }) + } } func TestAdaptResponsesClientToolsWithInheritedMapping_PromotesOmittedToolsDiscoveryIntoEffectiveDeclarations(t *testing.T) { diff --git a/backend/internal/service/openai_gateway_responses_client_tools_test.go b/backend/internal/service/openai_gateway_responses_client_tools_test.go index fa4866decf..1c9bdd6bcd 100644 --- a/backend/internal/service/openai_gateway_responses_client_tools_test.go +++ b/backend/internal/service/openai_gateway_responses_client_tools_test.go @@ -250,6 +250,34 @@ func TestOpenAIPassthroughAPIKeyRestoresClientToolsNonStreaming(t *testing.T) { require.Equal(t, "*** Begin Patch", gjson.Get(recorder.Body.String(), "output.1.input").String()) } +func TestOpenAIPassthroughAPIKeyPreservesCustomToolOutputContentParts(t *testing.T) { + gin.SetMode(gin.TestMode) + body := []byte(`{"model":"gpt-5.4","stream":false,"tools":[{"type":"custom","name":"exec"}],"input":[{"type":"custom_tool_call_output","call_id":"call_1","output":[{"type":"input_text","text":"result"},{"type":"input_file","file_id":"file_123"}]}]}`) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_tools","status":"completed","output":[],"usage":{}}`)), + }} + svc := openAIClientToolsTestService(upstream) + account := &Account{ID: 6240, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "test-key"}} + + result, err := svc.forwardOpenAIPassthrough(context.Background(), c, account, body, body, "gpt-5.4", false, nil, false, time.Now()) + + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, "function_call_output", gjson.GetBytes(upstream.lastBody, "input.0.type").String()) + output := gjson.GetBytes(upstream.lastBody, "input.0.output") + require.True(t, output.IsArray(), "native Responses content parts must reach the upstream as an array") + require.Equal(t, "input_text", output.Get("0.type").String()) + require.Equal(t, "result", output.Get("0.text").String()) + require.Equal(t, "input_file", output.Get("1.type").String()) + require.Equal(t, "file_123", output.Get("1.file_id").String()) +} + func TestOpenAIPassthroughAPIKeyRestoresClientToolsStreaming(t *testing.T) { gin.SetMode(gin.TestMode) body := openAIClientToolsRequest(true) From d881bfc0de7c38ae897ffef8d61919ad54a4038f Mon Sep 17 00:00:00 2001 From: feeeei Date: Wed, 26 Aug 2026 19:50:11 +0800 Subject: [PATCH 06/10] =?UTF-8?q?fix(openai):=20=E9=81=BF=E5=85=8D=20pool?= =?UTF-8?q?=20=E4=B8=A4=E8=B7=B3=E9=87=8D=E5=A4=8D=E8=AE=A1=E7=AE=97=20sys?= =?UTF-8?q?tem=20=E6=8F=90=E7=A4=BA=E8=AF=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/openai_gateway_forward.go | 20 +++++++++- .../service/openai_oauth_passthrough_test.go | 38 +++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 9a4c2389e3..4f2b102140 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -472,13 +472,29 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if decodeErr != nil { return nil, decodeErr } + // Responses OAuth 与 Chat 兼容入口保持一致：纯文本 system 可以无损提升后删除， + // JSON object 模式仍需在 input 中保留 JSON 指令供上游兼容校验。 + omitPromotedSystemMessages := !strings.EqualFold( + strings.TrimSpace(gjson.GetBytes(body, "text.format.type").String()), + "json_object", + ) codexResult := codexTransformResult{} if compatMessagesBridge { - codexResult = applyCodexOAuthTransformWithOptions(decoded, codexOAuthTransformOptions{IsCodexCLI: isCodexCLI, IsCompact: isCompactRequest, SkipDefaultInstructions: true, PreserveToolCallIDs: true}) + codexResult = applyCodexOAuthTransformWithOptions(decoded, codexOAuthTransformOptions{ + IsCodexCLI: isCodexCLI, + IsCompact: isCompactRequest, + SkipDefaultInstructions: true, + PreserveToolCallIDs: true, + OmitPromotedSystemMessagesFromInput: omitPromotedSystemMessages, + }) ensureCodexOAuthInstructionsField(decoded) markDecodedModified() } else { - codexResult = applyCodexOAuthTransform(decoded, isCodexCLI, isCompactRequest) + codexResult = applyCodexOAuthTransformWithOptions(decoded, codexOAuthTransformOptions{ + IsCodexCLI: isCodexCLI, + IsCompact: isCompactRequest, + OmitPromotedSystemMessagesFromInput: omitPromotedSystemMessages, + }) } if codexResult.Error != nil { c.JSON(http.StatusBadRequest, gin.H{"error": gin.H{"type": "invalid_request_error", "message": codexResult.Error.Error()}}) diff --git a/backend/internal/service/openai_oauth_passthrough_test.go b/backend/internal/service/openai_oauth_passthrough_test.go index fb9a6fee27..fa750fa5b5 100644 --- a/backend/internal/service/openai_oauth_passthrough_test.go +++ b/backend/internal/service/openai_oauth_passthrough_test.go @@ -132,6 +132,44 @@ func TestOpenAIGatewayService_ResponsesUnknownModelDoesNotFallbackToGPT54(t *tes require.True(t, rec.Code >= http.StatusBadRequest) } +func TestOpenAIGatewayService_OAuthResponsesPromotesSystemMessageWithoutDuplication(t *testing.T) { + gin.SetMode(gin.TestMode) + + const systemPrompt = "Unique system prefix for Responses token accounting." + const existingInstructions = "Existing instructions." + body := []byte(`{"model":"gpt-5.4","stream":false,"instructions":"` + existingInstructions + `","input":[{"role":"system","content":"` + systemPrompt + `"},{"role":"user","content":"hello"}]}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{err: errors.New("stop after capture")} + svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream} + account := &Account{ + ID: 124, + Name: "openai-oauth", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Concurrency: 1, + Credentials: map[string]any{ + "access_token": "oauth-token", + "chatgpt_account_id": "chatgpt-acc", + }, + Status: StatusActive, + Schedulable: true, + } + + result, err := svc.Forward(context.Background(), c, account, body) + + require.Error(t, err) + require.Nil(t, result) + require.NotEmpty(t, upstream.lastBody) + require.Equal(t, systemPrompt+"\n\n"+existingInstructions, gjson.GetBytes(upstream.lastBody, "instructions").String()) + require.Equal(t, int64(1), gjson.GetBytes(upstream.lastBody, "input.#").Int()) + require.Equal(t, "user", gjson.GetBytes(upstream.lastBody, "input.0.role").String()) + require.Equal(t, 1, strings.Count(string(upstream.lastBody), systemPrompt)) +} + func TestOpenAIGatewayService_NativeResponsesBodyModificationPreservesHTMLChars(t *testing.T) { gin.SetMode(gin.TestMode) From d077002ebaec8f5fde8383f7a892d0b054f658f5 Mon Sep 17 00:00:00 2001 From: li Date: Thu, 27 Aug 2026 13:36:50 +0800 Subject: [PATCH 07/10] =?UTF-8?q?fix(openai):=20=E5=9B=BE=E7=89=87?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E5=86=B7=E5=8D=B4=E4=B8=8D=E5=86=8D=E7=94=B1?= =?UTF-8?q?=E3=80=8C=E6=A8=A1=E5=9E=8B=E5=9B=9E=E6=96=87=E5=AD=97=E3=80=8D?= =?UTF-8?q?=E8=BF=99=E4=B8=80=E8=AF=B7=E6=B1=82=E7=BA=A7=E5=88=A4=E6=8D=AE?= =?UTF-8?q?=E8=A7=A6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v0.1.181 起，/v1/images/generations 只要上游回了文字而没回图，就给账号写 30 分钟 openai:image_generation 模型级冷却。该判据由 openAIImagesTextFallbackErrorForText 从模型的自然语言输出**推断**而来，只说明这个 prompt 这一轮模型选择了说话—— 仓库自己的既有用例 fixture 就是一句「Here's a polished image prompt for your request.」。 更糟的是同一个错误被判为可重试（502 → IsOpenAIImagesRetryableUpstreamError）并驱动 newOpenAIAccountFailoverError：一次闲聊回复会被逐个账号重试，沿途把每个账号都冷却掉， 表现为「开启后没多久账号就全部被限制」。 改为只在上游 error 帧点名该状态时才写冷却；网关自行推断的判据打上 SynthesizedFromModelText 标记并跳过。换号行为一字未动。 依据是仓库既有政策（alpha/search 路径原话）：工具端点的单次失败 「仍允许本次请求换号，但不修改任何账号状态」。 Fixes #6171 --- .../service/openai_images_responses.go | 35 +++- .../openai_images_tool_cooldown_test.go | 177 ++++++++++++++++++ .../ratelimit_service_openai_image_test.go | 57 +++++- 3 files changed, 266 insertions(+), 3 deletions(-) create mode 100644 backend/internal/service/openai_images_tool_cooldown_test.go diff --git a/backend/internal/service/openai_images_responses.go b/backend/internal/service/openai_images_responses.go index 9374a6f7c7..bbfbd5affb 100644 --- a/backend/internal/service/openai_images_responses.go +++ b/backend/internal/service/openai_images_responses.go @@ -37,6 +37,13 @@ type OpenAIImagesUpstreamError struct { Message string Param string UpstreamRequestID string + + // SynthesizedFromModelText marks an error the gateway inferred from the + // model's plain-text output instead of reading it off a structured upstream + // error frame. Such a verdict describes this one turn ("the model answered + // with words instead of an image"), not the account — see + // shouldCoolOpenAIImagesToolForError. + SynthesizedFromModelText bool } func (e *OpenAIImagesUpstreamError) Error() string { @@ -711,6 +718,10 @@ func openAIImagesTextFallbackErrorForText(text string) *OpenAIImagesUpstreamErro ErrorType: "upstream_error", Code: "image_generation_unavailable", Message: "Upstream did not execute image generation", + // Inferred from the model's own words, not from an upstream error frame: + // good enough to fail this turn over to another account, not evidence that + // this account's image tool is down for the next 30 minutes. + SynthesizedFromModelText: true, } } @@ -1922,6 +1933,26 @@ const ( openAIImagesOAuthUnavailableReason = "openai_images_oauth_tool_unavailable" ) +// shouldCoolOpenAIImagesToolForError decides whether an image_generation_unavailable +// verdict is durable enough to park the account's image tool for +// openAIImagesOAuthUnavailableCooldown. +// +// Only an upstream error frame that names the condition qualifies. A verdict the +// gateway synthesized from the model's plain-text reply does not: it merely says +// this prompt produced words instead of an image, which is prompt-dependent and +// happens on healthy accounts. Writing a 30-minute account-level cooldown from it +// is doubly wrong because the very same error is classified retryable +// (IsOpenAIImagesRetryableUpstreamError: status >= 500) and drives +// newOpenAIAccountFailoverError — so one such reply walks the pool and cools every +// account the retry touches. +// +// This mirrors the rule the alpha/search path already states in words: a +// tool-endpoint failure "仍允许本次请求换号，但不修改任何账号状态" +// (see shouldApplyOpenAIAlphaSearchAccountErrorSideEffects). +func shouldCoolOpenAIImagesToolForError(upstreamErr *OpenAIImagesUpstreamError) bool { + return upstreamErr != nil && !upstreamErr.SynthesizedFromModelText +} + func (s *OpenAIGatewayService) coolOpenAIImagesOAuthTool(ctx context.Context, account *Account) { if s == nil || s.accountRepo == nil || account == nil || account.Platform != PlatformOpenAI { return @@ -2017,7 +2048,9 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError( responseBody := openAIImagesUpstreamErrorResponseBody(upstreamErr) if upstreamErr.Code == "image_generation_unavailable" { - s.coolOpenAIImagesOAuthTool(ctx, account) + if shouldCoolOpenAIImagesToolForError(upstreamErr) { + s.coolOpenAIImagesOAuthTool(ctx, account) + } if responseWritten { return err } diff --git a/backend/internal/service/openai_images_tool_cooldown_test.go b/backend/internal/service/openai_images_tool_cooldown_test.go new file mode 100644 index 0000000000..0f38c200d7 --- /dev/null +++ b/backend/internal/service/openai_images_tool_cooldown_test.go @@ -0,0 +1,177 @@ +//go:build unit + +package service + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// issue #6171：v0.1.181 起，/v1/images/generations 只要上游"回文字没回图"，账号就被 +// 写 30 分钟 openai:image_generation 模型级冷却。该判据是**请求级**的（这个 prompt +// 这一轮模型选择了说话），却被当成**账号级**能力失效；又因为同一个错误被判为 +// 可重试（502）并驱动 failover，一次闲聊回复会沿着号池逐个把账号冷却掉。 + +// countingModelRateLimitRepo 记录 SetModelRateLimit 调用，用于断言"没写账号状态"。 +type countingModelRateLimitRepo struct { + accountRepoStub + calls int + scopes []string +} + +func (r *countingModelRateLimitRepo) SetModelRateLimit(_ context.Context, _ int64, scope string, _ time.Time, _ ...string) error { + r.calls++ + r.scopes = append(r.scopes, scope) + return nil +} + +func newImagesCooldownContext(t *testing.T) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + return c, rec +} + +func imagesCooldownAccount() *Account { + return &Account{ID: 77, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Name: "img-oauth"} +} + +func TestShouldCoolOpenAIImagesToolForError(t *testing.T) { + cases := []struct { + name string + err *OpenAIImagesUpstreamError + want bool + }{ + { + name: "nil_error", + err: nil, + want: false, + }, + { + // 网关从模型文字里推断出来的判据：只说明这一轮没出图。 + name: "synthesized_from_model_text", + err: &OpenAIImagesUpstreamError{ + StatusCode: http.StatusBadGateway, + Code: "image_generation_unavailable", + SynthesizedFromModelText: true, + }, + want: false, + }, + { + // 上游自己在 error 帧里点名该状态：这才是账号级证据，保持冷却。 + name: "structured_upstream_error_frame", + err: &OpenAIImagesUpstreamError{ + StatusCode: http.StatusBadGateway, + Code: "image_generation_unavailable", + }, + want: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, shouldCoolOpenAIImagesToolForError(tc.err)) + }) + } +} + +// 主复现：文字兜底判据不得写账号级冷却。 +func TestHandleOpenAIImagesOAuthResponseError_TextFallbackDoesNotCoolAccount(t *testing.T) { + c, _ := newImagesCooldownContext(t) + repo := &countingModelRateLimitRepo{} + svc := &OpenAIGatewayService{accountRepo: repo} + account := imagesCooldownAccount() + + upstreamErr := openAIImagesTextFallbackErrorForText("Here's a polished image prompt for your request.") + require.NotNil(t, upstreamErr) + require.Equal(t, "image_generation_unavailable", upstreamErr.Code) + + err := svc.handleOpenAIImagesOAuthResponseError( + context.Background(), c, account, "gpt-image-2", "https://upstream.example/v1/responses", + &http.Response{StatusCode: http.StatusOK, Header: http.Header{}}, + OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c), upstreamErr, + ) + + require.Zero(t, repo.calls, "模型闲聊不构成账号级证据，不得写 30 分钟冷却") + + // 换号行为必须原样保留：本 PR 只撤销账号状态写入，不动 failover。 + var failover *UpstreamFailoverError + require.True(t, errors.As(err, &failover), "仍应触发换号，got %T", err) +} + +// 对照不变式：上游 error 帧点名该状态时仍然冷却，否则等于把功能整个废掉。 +func TestHandleOpenAIImagesOAuthResponseError_StructuredUnavailableStillCoolsAccount(t *testing.T) { + c, _ := newImagesCooldownContext(t) + repo := &countingModelRateLimitRepo{} + svc := &OpenAIGatewayService{accountRepo: repo} + account := imagesCooldownAccount() + + upstreamErr := &OpenAIImagesUpstreamError{ + StatusCode: http.StatusBadGateway, + ErrorType: "upstream_error", + Code: "image_generation_unavailable", + Message: "image generation tool is not available for this account", + } + + _ = svc.handleOpenAIImagesOAuthResponseError( + context.Background(), c, account, "gpt-image-2", "https://upstream.example/v1/responses", + &http.Response{StatusCode: http.StatusOK, Header: http.Header{}}, + OpenAIImagesJSONKeepaliveAdjustedWrittenSize(c), upstreamErr, + ) + + require.Equal(t, 1, repo.calls, "结构化上游证据仍须写冷却") + require.Equal(t, []string{openAIImageGenerationRateLimitKey}, repo.scopes) +} + +// 标记必须打在文字兜底的两个入口上，且不影响违规拦截分支的判定。 +func TestOpenAIImagesTextFallback_MarksSynthesizedVerdicts(t *testing.T) { + t.Run("plain_text_reply_is_synthesized", func(t *testing.T) { + err := openAIImagesTextFallbackErrorForText("Here's a polished image prompt for your request.") + require.NotNil(t, err) + require.True(t, err.SynthesizedFromModelText) + require.Equal(t, "image_generation_unavailable", err.Code) + require.Equal(t, http.StatusBadGateway, err.StatusCode) + }) + + t.Run("body_entrypoint_is_synthesized", func(t *testing.T) { + body := []byte("event: response.completed\n" + + `data: {"type":"response.completed","response":{"id":"r","status":"completed",` + + `"output":[{"type":"message","content":[{"type":"output_text","text":"I drafted a prompt for you."}]}]}}` + + "\n\n") + err := openAIImagesTextFallbackError(body) + require.NotNil(t, err) + require.True(t, err.SynthesizedFromModelText) + }) + + t.Run("content_policy_branch_unchanged", func(t *testing.T) { + err := openAIImagesTextFallbackErrorForText("Blocked by our content policy.") + require.NotNil(t, err) + require.Equal(t, "content_policy_violation", err.Code) + require.Equal(t, http.StatusBadRequest, err.StatusCode) + // 该分支本来就不走冷却（Code 不匹配），标记与否都不改变行为； + // 断言它没有被顺手打标，避免语义漂移。 + require.False(t, err.SynthesizedFromModelText) + }) + + t.Run("empty_text_yields_no_error", func(t *testing.T) { + require.Nil(t, openAIImagesTextFallbackErrorForText(" ")) + }) +} + +// 级联的前提条件：该错误确实是可重试的，所以会带着"已写冷却"的副作用换号。 +// 这条用例把前提钉死，避免以后有人把 502 改成非重试后误以为本修复多余。 +func TestOpenAIImagesTextFallback_RemainsRetryableAndThusCascades(t *testing.T) { + err := openAIImagesTextFallbackErrorForText("Here's a polished image prompt for your request.") + require.NotNil(t, err) + require.True(t, IsOpenAIImagesRetryableUpstreamError(err), + "文字兜底判据是可重试的——正因如此，写账号冷却会沿号池级联") +} diff --git a/backend/internal/service/ratelimit_service_openai_image_test.go b/backend/internal/service/ratelimit_service_openai_image_test.go index 26714cf123..c08a43a98e 100644 --- a/backend/internal/service/ratelimit_service_openai_image_test.go +++ b/backend/internal/service/ratelimit_service_openai_image_test.go @@ -120,7 +120,11 @@ func TestOpenAIGatewayServiceForwardImages_ImageRateLimitReturnsFailoverAndCools require.Equal(t, openAIImageGenerationRateLimitKey, repo.modelRateLimitCalls[0].scope) } -func TestOpenAIGatewayServiceForwardImages_TextFallbackCoolsImageCapability(t *testing.T) { +// issue #6171：上游"回文字没回图"是**这一轮**的结果（模型选择了说话），不是账号能力 +// 失效。它同时被判为可重试（502）并驱动 failover，若还写 30 分钟账号级冷却，一次闲聊 +// 回复就会沿号池把每个被重试到的账号依次冷却掉。冷却仍保留给结构化上游证据，见 +// TestOpenAIGatewayServiceForwardImages_StructuredUnavailableCoolsImageCapability。 +func TestOpenAIGatewayServiceForwardImages_TextFallbackDoesNotCoolImageCapability(t *testing.T) { gin.SetMode(gin.TestMode) repo := &modelNotFoundAccountRepoStub{} body := []byte(`{"model":"gpt-image-2","prompt":"draw a cat"}`) @@ -154,7 +158,6 @@ func TestOpenAIGatewayServiceForwardImages_TextFallbackCoolsImageCapability(t *t }, } - before := time.Now() result, err := svc.ForwardImages(context.Background(), c, account, body, parsed, "") require.Nil(t, result) @@ -162,6 +165,56 @@ func TestOpenAIGatewayServiceForwardImages_TextFallbackCoolsImageCapability(t *t var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) require.False(t, failoverErr.RetryableOnSameAccount) + // 换号行为不变：该判据仍足以放弃本账号重试这一次请求…… + require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) + // ……但不再写任何账号级状态，否则重试会把冷却一路刷到整个号池。 + require.Empty(t, repo.modelRateLimitCalls, + "模型回文字只说明这一轮没出图，不构成账号 30 分钟不可用的证据") +} + +// 对照不变式：上游 error 帧点名 image_generation_unavailable 时仍写冷却， +// 保证 #6171 的修复没有把这项能力保护整个废掉。 +func TestOpenAIGatewayServiceForwardImages_StructuredUnavailableCoolsImageCapability(t *testing.T) { + gin.SetMode(gin.TestMode) + repo := &modelNotFoundAccountRepoStub{} + body := []byte(`{"model":"gpt-image-2","prompt":"draw a cat"}`) + upstreamSSE := "data: {\"type\":\"response.failed\",\"response\":{\"id\":\"r\",\"error\":" + + "{\"type\":\"upstream_error\",\"code\":\"image_generation_unavailable\"," + + "\"message\":\"image generation tool is not available for this account\"}}}\n\n" + + req := httptest.NewRequest(http.MethodPost, "/v1/images/generations", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = req + + svc := &OpenAIGatewayService{ + accountRepo: repo, + httpUpstream: &httpUpstreamRecorder{ + resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(upstreamSSE)), + }, + }, + } + parsed, err := svc.ParseOpenAIImagesRequest(c, body) + require.NoError(t, err) + account := &Account{ + ID: 206, + Name: "openai-oauth", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "access_token": "token-123", + }, + } + + before := time.Now() + result, err := svc.ForwardImages(context.Background(), c, account, body, parsed, "") + + require.Nil(t, result) + require.Error(t, err) require.Len(t, repo.modelRateLimitCalls, 1) call := repo.modelRateLimitCalls[0] require.Equal(t, account.ID, call.accountID) From 00b7c855cc84440e91f83787f0c63c920735b0a4 Mon Sep 17 00:00:00 2001 From: zhaotao1 Date: Thu, 27 Aug 2026 15:22:43 +0800 Subject: [PATCH 08/10] fix(openai): preserve declared API-key namespace calls --- .../service/openai_gateway_forward.go | 2 +- .../service/openai_responses_namespace.go | 31 ++++++++++++++++--- ...openai_responses_namespace_forward_test.go | 23 ++++++++++++++ .../openai_responses_namespace_test.go | 12 +++++-- 4 files changed, 60 insertions(+), 8 deletions(-) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 9a4c2389e3..9d4a8ef3e0 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -121,7 +121,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } if shouldStripOpenAIResponsesInputNamespaces(account, wsDecision.Transport, passthroughEnabled) { keepToolCallNamespaces := shouldKeepOpenAIResponsesToolCallNamespaces( - account, wsDecision.Transport, passthroughEnabled, compactPath, + account, wsDecision.Transport, passthroughEnabled, compactPath, body, ) body, err = stripOpenAIResponsesInputNamespaces(body, keepToolCallNamespaces) if err != nil { diff --git a/backend/internal/service/openai_responses_namespace.go b/backend/internal/service/openai_responses_namespace.go index f4fd47b13b..fdb0212402 100644 --- a/backend/internal/service/openai_responses_namespace.go +++ b/backend/internal/service/openai_responses_namespace.go @@ -77,25 +77,48 @@ func shouldStripOpenAIResponsesInputNamespaces(account *Account, transport OpenA // 故 OAuth 非 compact 请求必须保留。 // - compact 端点的 schema 不含该字段，携带即 400 `Unknown parameter: // input[N].namespace`（issue #4761 正文），故 compact 一律清理。 -// - API Key 出口是标准 Responses API（api.openai.com 或自定义 base_url），同样 -// 不认识该字段，维持全量清理；否则只能退化成 -// openai_responses_rejected_field_retry 的逐项删除，6 次上限根本盖不住长历史。 +// - API Key 出口默认按标准 Responses API 处理并清理该字段；但当请求本身声明 +// namespace 工具时，上游显然使用了 namespace 扩展，此时必须保留调用项上的 +// namespace，否则声明与历史调用会失配并触发 Missing namespace。 // - 摊平模式下调用项已被改写成平名，残留 namespace 指向的声明已不存在，一律清理。 func shouldKeepOpenAIResponsesToolCallNamespaces( account *Account, transport OpenAIUpstreamTransport, passthroughEnabled bool, compactPath bool, + body []byte, ) bool { - if account == nil || !account.IsOpenAIOAuthLike() { + if account == nil { return false } if compactPath { return false } + if account.IsOpenAIApiKey() { + return hasOpenAIResponsesNamespaceToolDeclaration(body) + } + if !account.IsOpenAIOAuthLike() { + return false + } return !shouldFlattenOpenAIResponsesNamespaces(account, transport, passthroughEnabled, compactPath) } +func hasOpenAIResponsesNamespaceToolDeclaration(body []byte) bool { + tools := gjson.GetBytes(body, "tools") + if !tools.IsArray() { + return false + } + found := false + tools.ForEach(func(_, tool gjson.Result) bool { + if strings.EqualFold(strings.TrimSpace(tool.Get("type").String()), "namespace") { + found = true + return false + } + return true + }) + return found +} + // openAIResponsesToolCallItemTypes 是携带 namespace 的调用项类型集合。与 // removeOpenAIResponsesRejectedNamespaceAtIndex 的反应式白名单保持一致；codex-rs // protocol/src/models.rs 中只有 FunctionCall 与 CustomToolCall 序列化 namespace， diff --git a/backend/internal/service/openai_responses_namespace_forward_test.go b/backend/internal/service/openai_responses_namespace_forward_test.go index e99d0199fd..9ef4f370b4 100644 --- a/backend/internal/service/openai_responses_namespace_forward_test.go +++ b/backend/internal/service/openai_responses_namespace_forward_test.go @@ -66,6 +66,29 @@ func TestOpenAIGatewayService_OAuthPreservesCodexNamespaceTools(t *testing.T) { require.Empty(t, openAIResponsesNamespaceNames(c)) } +// API Key 自定义上游若接受 namespace 工具声明，也要求历史 function_call 原样携带 +// namespace。声明仍为命名空间工具却清掉调用项字段，会触发 Missing namespace。 +func TestOpenAIGatewayService_APIKeyPreservesDeclaredNamespaceToolCalls(t *testing.T) { + body := []byte(codexNamespaceRequestBody) + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + newOpenAIRejectedFieldTestResponse(http.StatusOK, namespaceForwardOKResponse), + }} + c := newOpenAIRejectedFieldTestContext(body) + + result, err := newOpenAIRejectedFieldTestService(upstream).Forward( + context.Background(), c, newOpenAIRejectedFieldTestAccount(), body, + ) + + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.bodies, 1) + forwarded := upstream.bodies[0] + + require.True(t, gjson.GetBytes(forwarded, `tools.#(type=="namespace")`).Exists()) + require.Equal(t, "collaboration", gjson.GetBytes(forwarded, "input.0.namespace").String()) + require.False(t, gjson.GetBytes(forwarded, "input.1.namespace").Exists()) +} + // compact 端点 schema 更窄：input[].namespace 会 400 Unknown parameter（issue #4761）， // 且没有证据表明它接受 namespace 工具声明。compact 只做历史摘要、不需要模型寻址工具， // 因此保持既有的摊平 + 全量清理行为，不随默认值翻转扩大风险面。 diff --git a/backend/internal/service/openai_responses_namespace_test.go b/backend/internal/service/openai_responses_namespace_test.go index 7a9b47b7f5..f2ed2d8901 100644 --- a/backend/internal/service/openai_responses_namespace_test.go +++ b/backend/internal/service/openai_responses_namespace_test.go @@ -78,6 +78,7 @@ func TestShouldKeepOpenAIResponsesToolCallNamespaces(t *testing.T) { transport OpenAIUpstreamTransport passthroughEnabled bool compactPath bool + body []byte want bool }{ // 上游按 namespace 解析历史调用，缺字段会 400 "Missing namespace for function_call"。 @@ -92,15 +93,20 @@ func TestShouldKeepOpenAIResponsesToolCallNamespaces(t *testing.T) { // WSv2 + compact 是唯一「不摊平但仍必须清理」的组合，钉住 compact 判定本身， // 使其不会被误当成可由 shouldFlatten 推导出的冗余分支。 {name: "oauth_compact_wsv2_strips", account: oauth, transport: OpenAIUpstreamTransportResponsesWebsocketV2, compactPath: true, want: false}, - // API Key 出口是标准 Responses API，不认识该字段。 - {name: "apikey_strips", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, want: false}, + // API Key 默认按标准 Responses API 清理；请求显式声明 namespace 工具时， + // 自定义上游需要原样接收对应的历史调用。 + {name: "apikey_without_namespace_tool_strips", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, want: false}, + {name: "apikey_with_namespace_tool_keeps", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, body: []byte(`{"tools":[{"type":"namespace","name":"mcp__codex_app","tools":[]}]}`), want: true}, + {name: "apikey_with_mixed_case_namespace_tool_keeps", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, body: []byte(`{"tools":[{"type":" Namespace ","name":"mcp__codex_app","tools":[]}]}`), want: true}, + {name: "apikey_function_tool_with_namespace_field_strips", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, body: []byte(`{"tools":[{"type":"function","name":"automation_update","namespace":"mcp__codex_app"}]}`), want: false}, + {name: "apikey_compact_with_namespace_tool_strips", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, compactPath: true, body: []byte(`{"tools":[{"type":"namespace","name":"mcp__codex_app","tools":[]}]}`), want: false}, {name: "setup_token_keeps", account: setupToken, transport: OpenAIUpstreamTransportHTTPSSE, want: true}, {name: "nil_account", account: nil, transport: OpenAIUpstreamTransportHTTPSSE, want: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { require.Equal(t, tt.want, shouldKeepOpenAIResponsesToolCallNamespaces( - tt.account, tt.transport, tt.passthroughEnabled, tt.compactPath, + tt.account, tt.transport, tt.passthroughEnabled, tt.compactPath, tt.body, )) }) } From 00efee43034167f7b9c2d6ce7c0931720ff09d29 Mon Sep 17 00:00:00 2001 From: kk <5426729+kk-418@users.noreply.github.com> Date: Thu, 27 Aug 2026 15:48:06 +0800 Subject: [PATCH 09/10] =?UTF-8?q?=F0=9F=90=9B=20fix(grok):=20=E4=B8=BA=20G?= =?UTF-8?q?rok=204.6=20=E5=B9=BF=E5=91=8A=20xhigh=20=E6=8E=A8=E7=90=86?= =?UTF-8?q?=E6=A1=A3=E4=BD=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Grok Build 的 /v1/models 原先给所有可配 reasoning 的 Grok 模型写死三档， 导致 grok-4.6 无法在 /model 里选出官方支持的 xhigh。转发与 Codex catalog 已按模型判断，现与同一条件对齐。 --- backend/internal/handler/gateway_handler.go | 6 ++- .../internal/handler/gateway_models_test.go | 41 +++++++++++++++---- .../service/openai_codex_models_service.go | 2 +- .../internal/service/openai_gateway_grok.go | 6 ++- 4 files changed, 43 insertions(+), 12 deletions(-) diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index d07cd57e13..02e84fdcae 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -1307,11 +1307,15 @@ func writeGrokModelsList(c *gin.Context, modelIDs []string) { if grokModelSupportsConfigurableReasoning(modelID) { item.SupportsReasoningEffort = true item.ReasoningEffort = "high" - item.ReasoningEfforts = []grokReasoningEffortOption{ + efforts := []grokReasoningEffortOption{ {Value: "low", Label: "Low"}, {Value: "medium", Label: "Medium"}, {Value: "high", Label: "High", Default: true}, } + if service.GrokSupportsXHighReasoningEffort(modelID) { + efforts = append(efforts, grokReasoningEffortOption{Value: "xhigh", Label: "xHigh"}) + } + item.ReasoningEfforts = efforts } models = append(models, item) } diff --git a/backend/internal/handler/gateway_models_test.go b/backend/internal/handler/gateway_models_test.go index 05069bf25a..8059611faa 100644 --- a/backend/internal/handler/gateway_models_test.go +++ b/backend/internal/handler/gateway_models_test.go @@ -364,9 +364,38 @@ func TestGatewayModels_GeminiGroupFallsBackToGeminiModels(t *testing.T) { } func TestGatewayModels_Grok45AdvertisesReasoningEffortForGrokBuild(t *testing.T) { + assertGrokGatewayReasoningEfforts(t, 4409, "grok-4.5", []gatewayReasoningEffortOptionForTest{ + {Value: "low", Label: "Low"}, + {Value: "medium", Label: "Medium"}, + {Value: "high", Label: "High", Default: true}, + }) +} + +func TestGatewayModels_Grok46AdvertisesXHighReasoningEffortForGrokBuild(t *testing.T) { + xhighEfforts := []gatewayReasoningEffortOptionForTest{ + {Value: "low", Label: "Low"}, + {Value: "medium", Label: "Medium"}, + {Value: "high", Label: "High", Default: true}, + {Value: "xhigh", Label: "xHigh"}, + } + tests := []struct { + groupID int64 + model string + }{ + {groupID: 4410, model: "grok-4.6"}, + {groupID: 4411, model: "grok-4.6-latest"}, + } + for _, tt := range tests { + t.Run(tt.model, func(t *testing.T) { + assertGrokGatewayReasoningEfforts(t, tt.groupID, tt.model, xhighEfforts) + }) + } +} + +func assertGrokGatewayReasoningEfforts(t *testing.T, groupID int64, modelID string, want []gatewayReasoningEffortOptionForTest) { + t.Helper() gin.SetMode(gin.TestMode) - groupID := int64(4409) h := newGatewayModelsHandlerForTest( &gatewayModelsAccountRepoStub{ byGroup: map[int64][]service.Account{ @@ -375,7 +404,7 @@ func TestGatewayModels_Grok45AdvertisesReasoningEffortForGrokBuild(t *testing.T) ID: 1, Platform: service.PlatformGrok, Credentials: map[string]any{ - "model_mapping": map[string]any{"grok-4.5": "grok-4.5"}, + "model_mapping": map[string]any{modelID: modelID}, }, }, }, @@ -397,14 +426,10 @@ func TestGatewayModels_Grok45AdvertisesReasoningEffortForGrokBuild(t *testing.T) require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got)) require.Len(t, got.Data, 1) model := got.Data[0] - require.Equal(t, "grok-4.5", model.ID) + require.Equal(t, modelID, model.ID) require.True(t, model.SupportsReasoningEffort) require.Equal(t, "high", model.ReasoningEffort) - require.Equal(t, []gatewayReasoningEffortOptionForTest{ - {Value: "low", Label: "Low"}, - {Value: "medium", Label: "Medium"}, - {Value: "high", Label: "High", Default: true}, - }, model.ReasoningEfforts) + require.Equal(t, want, model.ReasoningEfforts) } func TestGatewayModels_GeminiGroupFiltersMappedModelsByPlatform(t *testing.T) { diff --git a/backend/internal/service/openai_codex_models_service.go b/backend/internal/service/openai_codex_models_service.go index 95be9ebbaf..b55549e753 100644 --- a/backend/internal/service/openai_codex_models_service.go +++ b/backend/internal/service/openai_codex_models_service.go @@ -477,7 +477,7 @@ func configuredCodexGrokReasoningLevels(modelID string) []configuredCodexReasoni {Effort: "medium", Description: "Balanced reasoning for most coding tasks"}, {Effort: "high", Description: "Greater reasoning depth for coding and agent tasks"}, } - if grokSupportsXHighReasoningEffort(modelID) { + if GrokSupportsXHighReasoningEffort(modelID) { levels = append(levels, configuredCodexReasoningLevel{ Effort: "xhigh", Description: "Extra-high reasoning depth for difficult tasks", diff --git a/backend/internal/service/openai_gateway_grok.go b/backend/internal/service/openai_gateway_grok.go index 21c7342356..a99f37ac2a 100644 --- a/backend/internal/service/openai_gateway_grok.go +++ b/backend/internal/service/openai_gateway_grok.go @@ -723,7 +723,7 @@ func normalizeGrokReasoningEffortValue(raw, model string) (string, bool) { case "minimal": return "low", true case "xhigh", "extrahigh": - if grokSupportsXHighReasoningEffort(model) { + if GrokSupportsXHighReasoningEffort(model) { return "xhigh", true } return "high", true @@ -734,7 +734,9 @@ func normalizeGrokReasoningEffortValue(raw, model string) (string, bool) { } } -func grokSupportsXHighReasoningEffort(model string) bool { +// GrokSupportsXHighReasoningEffort reports whether the model advertises and +// forwards the xhigh reasoning effort (Grok 4.6 and its undated alias). +func GrokSupportsXHighReasoningEffort(model string) bool { model = strings.ToLower(xai.StripGrokProviderPrefix(strings.TrimSpace(model))) return model == "grok-4.6" || model == "grok-4.6-latest" } From e6ea7b9af0fb560ec108a523cf87dd4194f19bbe Mon Sep 17 00:00:00 2001 From: Ryan Chou Date: Fri, 7 Aug 2026 20:46:25 +0800 Subject: [PATCH 10/10] =?UTF-8?q?fix(openai):=20=E5=9B=BE=E5=83=8F?= =?UTF-8?q?=E8=83=BD=E5=8A=9B=E4=B8=A2=E5=A4=B1=E6=97=B6=E5=86=B7=E5=8D=B4?= =?UTF-8?q?=20image=20=E8=B0=83=E5=BA=A6=EF=BC=8C=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E5=8F=8D=E5=A4=8D=E9=80=89=E4=B8=AD=E5=9D=8F=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OAuth 账号被上游收走 image_generation 能力后，/v1/images/* 稳定返回同一句 400 （Tool choice 'image_generation' not found in 'tools' parameter），但网关不做 任何处置，坏号持续被排程选中（报告者实测连续 17 次同 body 400、窗口内 0 成功、 持续 3.5 小时，直到人工停调度）。 RateLimitService 的 400 分支是白名单式，只认 organization has been disabled / credit balance / identity verification is required 三句，其余 400 明确不处理； 唯一能兜任意 400 的 tryTempUnschedulable 又要求账号开启 temp-unschedulable 开关，OAuth 账号默认没开。两条路都不命中，shouldDisable 恒为 false，也没有任何 model 作用域冷却。 修复收窄到 sub2api 自建 body 的 images 管线：buildOpenAIImagesResponsesRequest 产出的 body 必定同时带 tool_choice 与匹配的 image_generation tool，这条路上出现 该 400 只可能是上游收了账号能力。forwardOpenAIImagesOAuth 在 upstreamCtx 上打 自建标记（覆盖其下游全部三个错误分支），共用错误入口读到标记才把该 400 判为能力 丢失，对 openai:image_generation 作用域冷却 30 分钟（对齐 upstreamModelNotFound Cooldown），不禁用整颗账号——账号对文本请求依然健康。 未采用 issue 建议①的全局签名匹配：handleOpenAIAccountUpstreamError 同时服务 passthrough 路径，客户端可自带「tool_choice 指 image_generation 但 tools 不带」 的请求触发一字不差的同一句 400，全局匹配等于任何 API 用户一句话即可禁用健康 账号。附 5 个测试，其中防毒杀用例在撤掉该闸门后会实测失败。建议②（连续 N 次 400 熔断）涉及跨请求状态与配置项，另议。 Fixes #4466 Co-Authored-By: Claude Opus 5 (1M context) --- .../openai_account_runtime_block_fastpath.go | 11 ++ .../service/openai_images_responses.go | 21 ++++ backend/internal/service/ratelimit_service.go | 40 ++++++ .../ratelimit_service_openai_image_test.go | 117 ++++++++++++++++++ 4 files changed, 189 insertions(+) diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index 1f402202ca..7b59c28fd4 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -133,6 +133,17 @@ func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Cont return false } + // Self-built images requests always carry a matching image_generation tool, so a + // "tool choice not found in 'tools'" 400 means upstream revoked this account's + // image capability. Gated on the self-built marker: passthrough clients control + // their own tools/tool_choice and could otherwise poison a healthy account. + if isOpenAIImagesSelfBuiltRequest(ctx) && isOpenAIImageCapabilityLossError(statusCode, responseBody) { + if s != nil && s.rateLimitService != nil { + _ = s.rateLimitService.HandleOpenAIImageCapabilityLoss(stateCtx, account, statusCode, responseBody) + } + return false + } + if s == nil || account == nil { return false } diff --git a/backend/internal/service/openai_images_responses.go b/backend/internal/service/openai_images_responses.go index 9374a6f7c7..0e31d2d935 100644 --- a/backend/internal/service/openai_images_responses.go +++ b/backend/internal/service/openai_images_responses.go @@ -328,6 +328,26 @@ func openAIImageUploadToDataURL(upload OpenAIImagesUpload) (string, error) { return "data:" + contentType + ";base64," + base64.StdEncoding.EncodeToString(upload.Data), nil } +// openAIImagesSelfBuiltRequestContextKey marks a request whose upstream body was +// fully constructed by buildOpenAIImagesResponsesRequest, i.e. tool_choice and the +// matching image_generation tool are always both present and never client-controlled. +type openAIImagesSelfBuiltRequestContextKey struct{} + +func withOpenAIImagesSelfBuiltRequest(ctx context.Context) context.Context { + if ctx == nil { + ctx = context.Background() + } + return context.WithValue(ctx, openAIImagesSelfBuiltRequestContextKey{}, true) +} + +func isOpenAIImagesSelfBuiltRequest(ctx context.Context) bool { + if ctx == nil { + return false + } + selfBuilt, _ := ctx.Value(openAIImagesSelfBuiltRequestContextKey{}).(bool) + return selfBuilt +} + func buildOpenAIImagesResponsesRequest(parsed *OpenAIImagesRequest, toolModel string) ([]byte, error) { if parsed == nil { return nil, fmt.Errorf("parsed images request is required") @@ -1775,6 +1795,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth( if err != nil { return nil, err } + upstreamCtx = withOpenAIImagesSelfBuiltRequest(upstreamCtx) upstreamReq, err := s.buildUpstreamRequest(upstreamCtx, c, account, responsesBody, token, true, parsed.StickySessionSeed(), false) if err != nil { return nil, err diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index ff8bfac268..4d6be56b06 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -75,6 +75,8 @@ const ( const ( openAIImageRateLimitDefaultCooldown = time.Minute openAIImageRateLimitReason = "openai_image_rate_limited" + openAIImageCapabilityLossCooldown = 30 * time.Minute + openAIImageCapabilityLossReason = "openai_image_capability_lost" ) var openAIImageTryAgainPattern = regexp.MustCompile(`(?i)try again in\s+([0-9]+(?:\.[0-9]+)?)\s*(ms|s|sec|secs|second|seconds|m|min|mins|minute|minutes)`) @@ -2190,6 +2192,44 @@ func (s *RateLimitService) HandleOpenAIImageRateLimit(ctx context.Context, accou return true } +func (s *RateLimitService) HandleOpenAIImageCapabilityLoss(ctx context.Context, account *Account, statusCode int, responseBody []byte) bool { + if s == nil || account == nil || s.accountRepo == nil { + return false + } + if account.Platform != PlatformOpenAI { + return false + } + if !account.ShouldHandleErrorCode(statusCode) { + slog.Info("openai_image_capability_loss_skipped_by_error_code_policy", "account_id", account.ID, "status_code", statusCode) + return false + } + if !isOpenAIImageCapabilityLossError(statusCode, responseBody) { + return false + } + + resetAt := time.Now().Add(openAIImageCapabilityLossCooldown) + if err := s.accountRepo.SetModelRateLimit(ctx, account.ID, openAIImageGenerationRateLimitKey, resetAt, openAIImageCapabilityLossReason); err != nil { + slog.Warn("openai_image_capability_loss_set_model_rate_limit_failed", "account_id", account.ID, "scope", openAIImageGenerationRateLimitKey, "error", err) + return true + } + slog.Info("openai_image_capability_lost", "account_id", account.ID, "scope", openAIImageGenerationRateLimitKey, "reset_at", resetAt, "reset_in", time.Until(resetAt).Truncate(time.Second)) + return true +} + +// isOpenAIImageCapabilityLossError reports whether upstream rejected the +// image_generation tool choice that sub2api itself put into the request body. +// Only meaningful for self-built images requests, where tools always carries a +// matching image_generation entry — upstream saying otherwise means the account +// lost the capability. +func isOpenAIImageCapabilityLossError(statusCode int, body []byte) bool { + if statusCode != http.StatusBadRequest || len(body) == 0 { + return false + } + lower := strings.ToLower(string(body)) + return strings.Contains(lower, "image_generation") && + strings.Contains(lower, "not found in 'tools' parameter") +} + func isOpenAIImageRateLimitError(statusCode int, body []byte) bool { if statusCode != http.StatusTooManyRequests || len(body) == 0 { return false diff --git a/backend/internal/service/ratelimit_service_openai_image_test.go b/backend/internal/service/ratelimit_service_openai_image_test.go index 26714cf123..f00143bc24 100644 --- a/backend/internal/service/ratelimit_service_openai_image_test.go +++ b/backend/internal/service/ratelimit_service_openai_image_test.go @@ -169,3 +169,120 @@ func TestOpenAIGatewayServiceForwardImages_TextFallbackCoolsImageCapability(t *t require.Equal(t, openAIImagesOAuthUnavailableReason, call.reason) require.WithinDuration(t, before.Add(openAIImagesOAuthUnavailableCooldown), call.resetAt, time.Second) } + +func TestOpenAIGatewayServiceForwardImages_CapabilityLossCoolsImageScope(t *testing.T) { + gin.SetMode(gin.TestMode) + repo := &modelNotFoundAccountRepoStub{} + body := []byte(`{"model":"gpt-image-2","prompt":"draw a cat"}`) + errorBody := `{"error":{"message":"Tool choice 'image_generation' not found in 'tools' parameter.","param":"tool_choice","type":"invalid_request_error"}}` + + req := httptest.NewRequest(http.MethodPost, "/v1/images/generations", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = req + + svc := &OpenAIGatewayService{ + rateLimitService: &RateLimitService{accountRepo: repo}, + httpUpstream: &httpUpstreamRecorder{ + resp: &http.Response{ + StatusCode: http.StatusBadRequest, + Header: http.Header{"X-Request-Id": []string{"req_img_capability_lost"}}, + Body: io.NopCloser(strings.NewReader(errorBody)), + }, + }, + } + parsed, err := svc.ParseOpenAIImagesRequest(c, body) + require.NoError(t, err) + account := &Account{ + ID: 205, + Name: "openai-oauth", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "access_token": "token-123", + }, + } + + before := time.Now() + result, err := svc.ForwardImages(context.Background(), c, account, body, parsed, "") + + require.Nil(t, result) + require.Error(t, err) + require.Len(t, repo.modelRateLimitCalls, 1) + call := repo.modelRateLimitCalls[0] + require.Equal(t, account.ID, call.accountID) + require.Equal(t, openAIImageGenerationRateLimitKey, call.scope) + require.Equal(t, openAIImageCapabilityLossReason, call.reason) + require.WithinDuration(t, before.Add(openAIImageCapabilityLossCooldown), call.resetAt, time.Second) +} + +func TestOpenAIGatewayServiceHandleUpstreamError_PassthroughCapabilityLossDoesNotCool(t *testing.T) { + repo := &modelNotFoundAccountRepoStub{} + svc := &OpenAIGatewayService{rateLimitService: &RateLimitService{accountRepo: repo}} + account := &Account{ID: 206, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + body := []byte(`{"error":{"message":"Tool choice 'image_generation' not found in 'tools' parameter.","param":"tool_choice","type":"invalid_request_error"}}`) + + disabled := svc.handleOpenAIAccountUpstreamError(context.Background(), account, http.StatusBadRequest, http.Header{}, body, "gpt-5.5") + + require.False(t, disabled) + require.Empty(t, repo.modelRateLimitCalls) + _, wholeAccountBlocked := svc.openaiAccountRuntimeBlockUntil.Load(account.ID) + require.False(t, wholeAccountBlocked) +} + +func TestRateLimitServiceHandleOpenAIImageCapabilityLoss_IgnoresGenericBadRequest(t *testing.T) { + repo := &modelNotFoundAccountRepoStub{} + svc := &RateLimitService{accountRepo: repo} + account := &Account{ID: 207, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + body := []byte(`{"error":{"message":"Invalid type for input[0].arguments"}}`) + + handled := svc.HandleOpenAIImageCapabilityLoss(context.Background(), account, http.StatusBadRequest, body) + + require.False(t, handled) + require.Empty(t, repo.modelRateLimitCalls) +} + +func TestRateLimitServiceHandleOpenAIImageCapabilityLoss_RespectsPlatformAndErrorCodePolicy(t *testing.T) { + body := []byte(`{"error":{"message":"Tool choice 'image_generation' not found in 'tools' parameter.","param":"tool_choice","type":"invalid_request_error"}}`) + + t.Run("non_openai_platform", func(t *testing.T) { + repo := &modelNotFoundAccountRepoStub{} + svc := &RateLimitService{accountRepo: repo} + account := &Account{ID: 208, Platform: PlatformAnthropic, Type: AccountTypeOAuth} + + handled := svc.HandleOpenAIImageCapabilityLoss(context.Background(), account, http.StatusBadRequest, body) + + require.False(t, handled) + require.Empty(t, repo.modelRateLimitCalls) + }) + + t.Run("custom_error_code_policy_excludes_400", func(t *testing.T) { + repo := &modelNotFoundAccountRepoStub{} + svc := &RateLimitService{accountRepo: repo} + account := &Account{ + ID: 209, + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Credentials: map[string]any{ + "custom_error_codes_enabled": true, + "custom_error_codes": []any{float64(http.StatusTooManyRequests)}, + }, + } + + require.False(t, account.ShouldHandleErrorCode(http.StatusBadRequest)) + handled := svc.HandleOpenAIImageCapabilityLoss(context.Background(), account, http.StatusBadRequest, body) + + require.False(t, handled) + require.Empty(t, repo.modelRateLimitCalls) + }) +} + +func TestIsOpenAIImageCapabilityLossError(t *testing.T) { + capabilityLossBody := []byte(`{"error":{"message":"Tool choice 'image_generation' not found in 'tools' parameter.","param":"tool_choice","type":"invalid_request_error"}}`) + genericBadRequestBody := []byte(`{"error":{"message":"Invalid type for input[0].arguments"}}`) + + require.True(t, isOpenAIImageCapabilityLossError(http.StatusBadRequest, capabilityLossBody)) + require.False(t, isOpenAIImageCapabilityLossError(http.StatusBadRequest, genericBadRequestBody)) + require.False(t, isOpenAIImageCapabilityLossError(http.StatusTooManyRequests, capabilityLossBody)) +}