fix(proxy): 首包前不可重试失败补齐 OpenAI 直连与入站 WS 两条路径

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"},

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

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) == "" {