From c9a2dfa4d1e5ecf17b300af85288858e3ec65e2e Mon Sep 17 00:00:00 2001 From: james-6-23 Date: Wed, 8 Jul 2026 22:15:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(proxy):=20=E9=A6=96=E5=8C=85=E5=89=8D?= =?UTF-8?q?=E4=B8=8D=E5=8F=AF=E9=87=8D=E8=AF=95=E5=A4=B1=E8=B4=A5=E8=A1=A5?= =?UTF-8?q?=E9=BD=90=20OpenAI=20=E7=9B=B4=E8=BF=9E=E4=B8=8E=E5=85=A5?= =?UTF-8?q?=E7=AB=99=20WS=20=E4=B8=A4=E6=9D=A1=E8=B7=AF=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #318/#332 只覆盖了 Responses/ChatCompletions 主流式路径,本次补齐: OpenAI Responses API 直连分支(IsOpenAIResponsesAPI): - 回调拦截首 token 前的 response.failed,收尾 flush 加 wroteAnyBody 守卫, 循环外覆盖 Content-Type 后按 outcome.logStatusCode 返回 JSON 错误, 与主路径行为一致(此前该分支把失败事件写进 200 流) 入站 WebSocket(responses_ws.go): - WS 升级后没有 HTTP 状态码可用,等价语义为:首 token 前不可重试的 response.failed 不透传原始失败帧,改写结构化 error 帧后按错误类别 用非正常 close code 关闭(429→1013,其余 4xx→1008,5xx→1011), 下游不再把失败会话当正常收尾 - 可重试失败不拦截:silent retry 开启时仍换号重试,关闭时按既有约定 原样透传失败帧(TestResponsesWebSocketSilentRetryDisabledRelaysRetryableFailure) 测试: - 新增 close code 映射单测 - 新增端到端 WS 测试:断言 error 帧 + 1008 close + 不换号重试 --- proxy/handler.go | 23 +++++- proxy/response_failed_billing_test.go | 112 ++++++++++++++++++++++++++ proxy/responses_ws.go | 37 +++++++++ 3 files changed, 171 insertions(+), 1 deletion(-) diff --git a/proxy/handler.go b/proxy/handler.go index fa856c8d..1c788f8c 100644 --- a/proxy/handler.go +++ b/proxy/handler.go @@ -1822,6 +1822,9 @@ func (h *Handler) Responses(c *gin.Context) { var readErr error var writeErr error wroteAnyBody := false + // 首 token 前收到不可重试的 response.failed 时置位:中止 SSE 转发、 + // 不做 transport flush(避免提前提交 200 header),循环外按真实错误码返回 JSON。 + abortedForHTTPError := false var imageLogInfo imageUsageLogInfo var terminalFailurePayload []byte @@ -1871,6 +1874,13 @@ func (h *Handler) Responses(c *gin.Context) { pendingFirstTokenEvents.Reset() return false } + // 首 token 前的 response.failed 不写进下游流:不可重试(如 context_length_exceeded) + // 或已达重试上限时,交由循环外按真实错误码返回,而不是 200 流让中转层误计费。 + if shouldReturnHTTPErrorForResponseFailed(eventType, ttftRecorded, wroteAnyBody, clientGone) { + pendingFirstTokenEvents.Reset() + abortedForHTTPError = true + return false + } if image, ok := extractImageFromOutputItemDone(data, logModel); ok { imageLogInfo = mergeImageUsageLogInfo(imageLogInfo, imageUsageLogInfoFromImage(image)) } @@ -1886,7 +1896,9 @@ func (h *Handler) Responses(c *gin.Context) { } return eventType != "response.completed" && eventType != "response.failed" }) - if writeErr == nil { + // 仅在真的写过 body 时才做收尾 flush:flusher.Flush 会先提交 HTTP 200 header, + // 零写入时提前 flush 会让循环外的 c.JSON(4xx) 失效(status 已定型为 200)。 + if writeErr == nil && wroteAnyBody { writeErr = streamWriter.Flush() } } else { @@ -1938,6 +1950,15 @@ func (h *Handler) Responses(c *gin.Context) { h.store.UnbindSessionAffinity(affinityKey, account.ID()) continue } + if isStream && abortedForHTTPError && !wroteAnyBody { + // 流式:首 token 前上游失败、未向下游写过任何内容,HTTP 200 header 尚未提交, + // 覆盖预设的 SSE Content-Type 后按真实错误码返回 JSON, + // 避免下游中转/计费方把它当成功并按预估 input token 计费(与回调内 reset 呼应)。 + c.Header("Content-Type", "application/json; charset=utf-8") + c.JSON(outcome.logStatusCode, gin.H{ + "error": gin.H{"message": outcome.failureMessage, "type": "upstream_error"}, + }) + } if !isStream && readErr != nil { c.JSON(http.StatusBadGateway, gin.H{ "error": gin.H{"message": "读取 OpenAI Responses 响应失败", "type": "upstream_error"}, diff --git a/proxy/response_failed_billing_test.go b/proxy/response_failed_billing_test.go index 0c79bdcc..15d75bc5 100644 --- a/proxy/response_failed_billing_test.go +++ b/proxy/response_failed_billing_test.go @@ -2,13 +2,19 @@ package proxy import ( "bytes" + "context" "io" "net/http" "net/http/httptest" "strings" "testing" + "time" + "github.com/codex2api/auth" + "github.com/codex2api/config" + "github.com/codex2api/database" "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" "github.com/tidwall/gjson" ) @@ -162,3 +168,109 @@ func TestStreamFirstTokenFailureWireBehavior(t *testing.T) { }) } } + +// 入站 WS 路径的等价修复:首 token 前不可重试的 response.failed 不透传原始失败帧, +// 改为 error 帧 + 按错误类别的非正常 close code(与 SSE 路径返回真实 4xx 对齐)。 +func TestResponsesWSCloseCodeForStatus(t *testing.T) { + cases := []struct { + status int + want int + }{ + {http.StatusTooManyRequests, websocket.CloseTryAgainLater}, + {http.StatusBadRequest, websocket.ClosePolicyViolation}, + {http.StatusUnauthorized, websocket.ClosePolicyViolation}, + {http.StatusForbidden, websocket.ClosePolicyViolation}, + {http.StatusPaymentRequired, websocket.ClosePolicyViolation}, + {http.StatusInternalServerError, websocket.CloseInternalServerErr}, + {http.StatusBadGateway, websocket.CloseInternalServerErr}, + {http.StatusOK, websocket.CloseInternalServerErr}, + } + for _, tc := range cases { + if got := responsesWSCloseCodeForStatus(tc.status); got != tc.want { + t.Errorf("responsesWSCloseCodeForStatus(%d) = %d, want %d", tc.status, got, tc.want) + } + } +} + +// 端到端:入站 WS 首 token 前收到不可重试的 response.failed(context_length_exceeded) +// 时,客户端应收到结构化 error 帧 + ClosePolicyViolation 关闭,而不是原始失败帧 + +// 正常收尾(后者会被下游中转/计费方当成一次成功会话)。且确定性客户端错误不换号重试。 +func TestResponsesWebSocketNonRetryableFailureReturnsErrorClose(t *testing.T) { + gin.SetMode(gin.TestMode) + + previousExec := WebsocketExecuteFunc + previousSettings := CurrentRuntimeSettings() + t.Cleanup(func() { + WebsocketExecuteFunc = previousExec + ApplyRuntimeSettings(previousSettings) + }) + nextSettings := previousSettings + nextSettings.CodexWSSilentRetry = true + nextSettings.CodexWSHideErrors = false + nextSettings.CodexWSSilentRetries = 2 + ApplyRuntimeSettings(nextSettings) + + attemptCh := make(chan int64, 4) + WebsocketExecuteFunc = func(ctx context.Context, account *auth.Account, requestBody []byte, sessionID string, proxyOverride string, apiKey string, deviceCfg *DeviceProfileConfig, headers http.Header, poolRouteKey string) (*http.Response, error) { + attemptCh <- account.ID() + sse := `data: {"type":"response.created"}` + "\n\n" + + `data: {"type":"response.failed","response":{"error":{"code":"context_length_exceeded","message":"input too long"}}}` + "\n\n" + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(sse)), + }, nil + } + + store := auth.NewStore(nil, nil, &database.SystemSettings{MaxConcurrency: 2, TestConcurrency: 1, TestModel: "gpt-5.4"}) + store.AddAccount(&auth.Account{DBID: 1, AccessToken: "at-1", PlanType: "pro", AccountID: "acct-1"}) + store.AddAccount(&auth.Account{DBID: 2, AccessToken: "at-2", PlanType: "pro", AccountID: "acct-2"}) + handler := NewHandler(store, nil, &config.Config{AllowAnonymousV1: true}, nil) + + router := gin.New() + handler.RegisterRoutes(router) + server := httptest.NewServer(router) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/v1/responses" + conn, resp, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + if resp != nil { + t.Fatalf("dial websocket failed: %v status=%d", err, resp.StatusCode) + } + t.Fatalf("dial websocket failed: %v", err) + } + defer conn.Close() + + if err := conn.WriteMessage(websocket.TextMessage, []byte(`{"model":"gpt-5.4","input":"hello"}`)); err != nil { + t.Fatalf("write request: %v", err) + } + + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + _, first, err := conn.ReadMessage() + if err != nil { + t.Fatalf("read error frame: %v", err) + } + if eventType := gjson.GetBytes(first, "type").String(); eventType != "error" { + t.Fatalf("first event type = %q, want \"error\" (原始 response.failed 帧不应透传) body=%s", eventType, first) + } + if !strings.Contains(string(first), "input too long") { + t.Fatalf("error frame should carry upstream message when hiding disabled: %s", first) + } + + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + if _, _, err := conn.ReadMessage(); !websocket.IsCloseError(err, websocket.ClosePolicyViolation) { + t.Fatalf("expected close %d (policy violation for deterministic 4xx), got err=%v", websocket.ClosePolicyViolation, err) + } + + select { + case <-attemptCh: + case <-time.After(time.Second): + t.Fatal("timed out waiting for first attempt") + } + select { + case got := <-attemptCh: + t.Fatalf("unexpected retry on account %d for non-retryable failure", got) + case <-time.After(100 * time.Millisecond): + } +} diff --git a/proxy/responses_ws.go b/proxy/responses_ws.go index fe050f36..5c1f79f6 100644 --- a/proxy/responses_ws.go +++ b/proxy/responses_ws.go @@ -499,6 +499,10 @@ func (h *Handler) streamResponsesWSUpstream( var imageLogInfo imageUsageLogInfo var terminalFailurePayload []byte wroteAnyBody := false + // 首 token 前收到不可重试的 response.failed 时置位:不把原始失败帧透传给客户端, + // 循环外改写 error 帧并按错误类别用非正常 close code 关闭, + // 让下游中转/计费方明确感知失败,而不是把它当成一次正常结束的会话。 + abortedForErrorClose := false pendingFirstTokenMessages := make([][]byte, 0, 4) pendingFirstTokenBytes := 0 @@ -564,6 +568,18 @@ func (h *Handler) streamResponsesWSUpstream( pendingFirstTokenBytes = 0 return false } + // 首 token 前的不可重试 response.failed(如 context_length_exceeded) + // 不透传原始失败帧:丢弃前导缓冲并提前结束读取,循环外按真实错误 + // 语义返回 error 帧 + 非正常 close code(与 SSE 路径返回 4xx 对齐)。 + // 可重试的失败不在此拦截:silent retry 开启时由上面的分支换号重试, + // 关闭时按既有约定原样透传失败帧。 + if shouldReturnHTTPErrorForResponseFailed(eventType, ttftRecorded, wroteAnyBody, clientGone) && + !responseFailedRetryable(terminalFailurePayload) { + pendingFirstTokenMessages = pendingFirstTokenMessages[:0] + pendingFirstTokenBytes = 0 + abortedForErrorClose = true + return false + } if len(pendingFirstTokenMessages) > 0 && !flushPendingFirstTokenMessages() { return false } @@ -675,6 +691,14 @@ func (h *Handler) streamResponsesWSUpstream( if writeErr != nil { return errResponsesWSClientGone } + if abortedForErrorClose && !wroteAnyBody { + // 首 token 前上游失败且未向客户端写过任何帧:发结构化 error 帧后按错误类别 + // 关闭连接,避免下游把"正常收尾的会话"当成功并按预估 input token 计费。 + apiErr := api.NewAPIError(api.ErrCodeUpstreamError, outcome.failureMessage, api.ErrorTypeUpstream) + clientErr := responsesWSClientUpstreamAPIError(apiErr, hideUpstreamErrors) + _ = writeResponsesWSError(conn, clientErr) + return newResponsesWSCloseError(responsesWSCloseCodeForStatus(outcome.logStatusCode), clientErr.Message, apiErr) + } if outcome.logStatusCode != http.StatusOK && hideUpstreamErrors && len(terminalFailurePayload) > 0 && !wroteAnyBody { apiErr := api.NewAPIError(api.ErrCodeUpstreamError, outcome.failureMessage, api.ErrorTypeUpstream) clientErr := responsesWSClientUpstreamAPIError(apiErr, true) @@ -820,6 +844,19 @@ func newResponsesWSCloseError(code int, reason string, err error) error { } } +// responsesWSCloseCodeForStatus 把上游失败的 HTTP 语义状态码映射为 WebSocket close code: +// 429 → 1013(稍后重试);其余 4xx 确定性客户端错误 → 1008(策略拒绝);5xx → 1011。 +func responsesWSCloseCodeForStatus(statusCode int) int { + switch { + case statusCode == http.StatusTooManyRequests: + return websocket.CloseTryAgainLater + case statusCode >= 400 && statusCode < 500: + return websocket.ClosePolicyViolation + default: + return websocket.CloseInternalServerErr + } +} + func responsesWSUpstreamAPIError(statusCode int, body []byte) *api.APIError { message := usageLogErrorMessage(statusCode, body) if strings.TrimSpace(message) == "" {