diff --git a/backend/internal/proxy/gateway.go b/backend/internal/proxy/gateway.go index e2550de..b825e6e 100644 --- a/backend/internal/proxy/gateway.go +++ b/backend/internal/proxy/gateway.go @@ -302,13 +302,25 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { g.channelSvc.RecordSuccess(ch.ID) // Stream or buffer response;tok 从上游响应(SSE usage 块或非流式 JSON)提取。 + // 上游可能返回 HTTP 200 但 body/SSE 内带 error(OpenRouter 超时等), + // 此时按失败记账(errCode 非空),非流式错误体以 502 返回给客户端。 var tok convert.TokenUsage + var errCode string if req.Stream { - tok = g.streamResponse(c, resp, req.Protocol, targetFormat, req.rawBuf) + tok, errCode = g.streamResponse(c, resp, req.Protocol, targetFormat, req.rawBuf) } else { - tok = g.bufferResponse(c, resp, req.Protocol, targetFormat, req.rawBuf) + tok, errCode, _ = g.bufferResponse(c, resp, req.Protocol, targetFormat, req.rawBuf) } resp.Body.Close() + if errCode != "" { + // 记账为失败(错误码),不产生费用;响应内容已由 buffer/stream 写出 + g.recordUsage(req, cand, ch, usage.Event{ + IsError: true, + ErrorCode: errCode, + LatencyMS: int(time.Since(start).Milliseconds()), + }, tok) + return + } // 成功记录:用量 + 定价计费。 g.recordUsage(req, cand, ch, usage.Event{ LatencyMS: int(time.Since(start).Milliseconds()), @@ -433,7 +445,9 @@ func (g *Gateway) setHeaders(req *http.Request, ch *store.Channel, apiKey string // streamResponse 流式响应:按 \n\n 分块零缓冲转发;跨协议时逐行转换。 // 返回从上游 SSE usage 块累计的 token 用量(按上游协议解析)。 // capture 非 nil 时把上游原始行累积进去(原始响应记录)。 -func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) convert.TokenUsage { +// 上游部分实现(如 OpenRouter)在超时时返回 HTTP 200 但 SSE data 内带 +// error 字段;检测到则返回错误码,供 Dispatch 按失败记账。 +func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) (convert.TokenUsage, string) { w := c.Writer c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") @@ -452,6 +466,7 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt // 同时喂入用量累计器(usage 块可能出现在任一事件)。 r := bufio.NewReaderSize(resp.Body, 32*1024) accum := convert.NewStreamUsageAccum() + errCode := "" for { buf := []byte{} for { @@ -463,14 +478,14 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt buf = append(buf, line...) if err == io.EOF { if len(buf) == 0 { - return accum.Usage() + return accum.Usage(), errCode } if !bytes.HasSuffix(buf, []byte("\n")) { buf = append(buf, '\n') } } else if err != nil { log.Printf("stream read error: %v", err) - return accum.Usage() + return accum.Usage(), errCode } if len(buf) >= 2 && bytes.HasSuffix(buf, []byte("\n\n")) { break @@ -485,6 +500,9 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt // 先解析用量(data: {...} 行),再决定转发内容。 for _, data := range sseDataPayloads(buf) { accum.Feed(data, upstreamProto) + if errCode == "" && streamChunkHasError(data) { + errCode = "upstream_stream_error" + } } out := buf @@ -495,7 +513,7 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt continue } if _, err := w.Write(out); err != nil { - return accum.Usage() // 客户端已断开 + return accum.Usage(), errCode // 客户端已断开 } if flusher != nil { flusher.Flush() @@ -505,11 +523,24 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt // 发完 [DONE] 后不关连接,继续读会阻塞到超时;据此主动收尾。 // responses 协议没有 [DONE],以 response.completed 事件收尾。 if streamTerminated(buf, upstreamProto) { - return accum.Usage() + return accum.Usage(), errCode } } } +// streamChunkHasError 判断一块 SSE data 载荷是否带 error 字段(OpenRouter 超时等)。 +func streamChunkHasError(data []byte) bool { + var m map[string]any + if json.Unmarshal(data, &m) != nil { + return false + } + if _, ok := m["error"]; ok { + return true + } + // responses 协议错误事件可能形如 {"type":"error",...} + return m["type"] == "error" +} + // streamTerminated 判断一块 SSE 是否为上游流的结束事件。 func streamTerminated(chunk []byte, proto string) bool { switch proto { @@ -546,11 +577,14 @@ func sseDataPayloads(chunk []byte) [][]byte { return out } -func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) convert.TokenUsage { +// bufferResponse 非流式响应:整体读取、可选转换后写回。 +// 返回 (用量, 错误码, 是否错误)。部分上游(如 OpenRouter)在超时时返回 +// HTTP 200 但 JSON 内含 error 字段,需要识别并让调用方按失败处理。 +func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) (convert.TokenUsage, string, bool) { body, err := io.ReadAll(resp.Body) if err != nil { g.writeError(c, http.StatusBadGateway, "failed to read response") - return convert.TokenUsage{} + return convert.TokenUsage{}, "", false } // 原始响应捕获(仅管理员+开关开启时启用)。 @@ -561,6 +595,9 @@ func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProt // 用量从上游原始响应体提取(先于转换,转换会改字段名)。 tok, _ := convert.ExtractUsageJSON(body, upstreamProto) + // HTTP 200 但带 error 字段(OpenRouter 超时 504 等):识别并转失败。 + errCode, isErr := bodyHasError(body) + out := body if upstreamProto != clientProto { if converted, cerr := convert.ConvertResponse(body, upstreamProto, clientProto); cerr == nil { @@ -573,8 +610,29 @@ func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProt // 直通:部分上游(如 OpenRouter)的 non-stream 响应在 JSON 前夹带空白/注释 out = convert.CleanJSON(body) } + // 上游错误体:用 502 返回,让客户端感知失败(不伪装成 200)。 + if isErr { + c.Data(http.StatusBadGateway, "application/json", out) + return tok, errCode, true + } c.Data(resp.StatusCode, "application/json", out) - return tok + return tok, errCode, false +} + +// bodyHasError 判断 JSON 响应体是否带 error 字段(openai 风格 {"error":{...}} 或 +// anthropic 风格 {"type":"error",...})。返回 (错误码, 是否错误)。找不到 JSON 返回 ("", false)。 +func bodyHasError(body []byte) (string, bool) { + var m map[string]any + if json.Unmarshal(bytes.TrimSpace(body), &m) != nil { + return "", false + } + if _, ok := m["error"]; ok { + return "upstream_error", true + } + if m["type"] == "error" { + return "upstream_error", true + } + return "", false } func (g *Gateway) writeError(c *gin.Context, status int, message string) {