M0-M4: 推倒重来基线(基建+用户/密钥/核心代理+前端+管理后台+三协议互转)

- 后端 Go+Gin+GORM: 配置(OT_ env)/SQLite/Postgres 双驱动、用户体系(argon2id+JWT access/refresh)、
  API Key(sk- 48位, 仅存 SHA-256 哈希)
- 代理网关: /v1/chat/completions、/v1/responses、/v1/messages、/v1/models;错误按客户端协议返回
- 三协议互转(convert 包): Chat↔Messages↔Responses 请求/响应 + 流式 SSE 逐事件转换(直通优先)
- 用量计费: 异步批量记账、余额扣减、balance_logs、usage_daily 日聚合
- 管理 API: 用户/渠道 CRUD+测试+模型导入/模型定价+绑定/统计/系统配置
- 前端 Vue3+TS+Tailwind(taste-skill 设计 tokens): Landing/登录注册/控制台/管理后台,
  自建组件+Phosphor 图标+自建 SVG 趋势图, 已过 web-design-guidelines 复查
- mock 上游: OpenAI+Anthropic 双协议模拟(含流式)

Co-Authored-By: Claude <noreply@anthropic.com>

server/internal/proxy/passthrough.go

@@ -43,24 +43,22 @@ func parseBody(c *gin.Context) (*bodyReq, []byte, error) {
 	return br, body, nil
 }
 
-// upstreamURL 组装上游地址：base_url + 客户端路径（/v1/chat/completions 等）。
+// upstreamURL 组装上游地址：base_url + 路径。
 func upstreamURL(ch *store.Channel, path string) string {
-	base := strings.TrimRight(ch.BaseURL, "/")
-	return base + path
+	return strings.TrimRight(ch.BaseURL, "/") + path
 }
 
-// doPassthrough 通用直通：替换 Authorization 为渠道密钥，转发请求。
-// convert 回调用于改写请求体（M1 直通为原样；M3 转换时改写）。
-func (g *Gateway) doPassthrough(c *gin.Context, ch *store.Channel, path string, body []byte, stream bool, outUsage func(usageRaw json.RawMessage)) {
+// doProxy 通用代理：替换 Authorization 为渠道密钥，转发请求；按 plan 决定路径与转换。
+func (g *Gateway) doProxy(c *gin.Context, ch *store.Channel, plan *upstreamPlan, stream bool, sink *usageSink) {
 	upKey, err := g.ch.UpstreamKey(ch)
 	if err != nil {
-		openAIError(c, http.StatusInternalServerError, "channel_error", "failed to decrypt channel key")
+		apiError(c, http.StatusInternalServerError, "channel_error", "failed to decrypt channel key")
 		return
 	}
 
-	upBody := body
-	// 流式 chat：注入 stream_options.include_usage，保证末块带 usage（OpenAI 行为）
-	if stream && path == "/v1/chat/completions" && !bytes.Contains(upBody, []byte(`"include_usage"`)) {
+	upBody := plan.body
+	// 直通 chat 流式：注入 stream_options.include_usage，保证末块带 usage（OpenAI 行为）
+	if stream && plan.path == "/v1/chat/completions" && plan.lineConv == nil && !bytes.Contains(upBody, []byte(`"include_usage"`)) {
 		var m map[string]any
 		if json.Unmarshal(upBody, &m) == nil {
 			m["stream_options"] = map[string]any{"include_usage": true}
@@ -72,9 +70,9 @@ func (g *Gateway) doPassthrough(c *gin.Context, ch *store.Channel, path string,
 
 	ctx, cancel := context.WithTimeout(c.Request.Context(), time.Duration(ch.TimeoutMS)*time.Millisecond)
 	defer cancel()
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, upstreamURL(ch, path), bytes.NewReader(upBody))
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, upstreamURL(ch, plan.path), bytes.NewReader(upBody))
 	if err != nil {
-		openAIError(c, http.StatusInternalServerError, "internal_error", "failed to build upstream request")
+		apiError(c, http.StatusInternalServerError, "internal_error", "failed to build upstream request")
 		return
 	}
 	req.Header.Set("Content-Type", "application/json")
@@ -83,6 +81,9 @@ func (g *Gateway) doPassthrough(c *gin.Context, ch *store.Channel, path string,
 	if ua := c.GetHeader("User-Agent"); ua != "" {
 		req.Header.Set("User-Agent", ua)
 	}
+	if plan.path == "/v1/messages" {
+		req.Header.Set("anthropic-version", "2023-06-01")
+	}
 	// 透传 OpenAI 生态请求头（组织/项目等）
 	for _, h := range []string{"OpenAI-Organization", "OpenAI-Project", "OpenAI-Beta"} {
 		if v := c.GetHeader(h); v != "" {
@@ -99,55 +100,57 @@ func (g *Gateway) doPassthrough(c *gin.Context, ch *store.Channel, path string,
 			status = http.StatusGatewayTimeout
 			msg = "Upstream request timed out"
 		}
-		openAIError(c, status, "upstream_error", msg)
+		apiError(c, status, "upstream_error", msg)
 		g.recordError(c, ch, nil, start, "upstream_error")
 		return
 	}
 	defer resp.Body.Close()
 
-	// 非 2xx：透传上游错误体（OpenAI 格式），并记录 error 用量
+	// 非 2xx：透传上游错误体，并记录 error 用量
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 		errBody, _ := io.ReadAll(resp.Body)
 		status := resp.StatusCode
-		// 上游 5xx → 网关 502/504（重试逻辑 M4）
 		if status >= 500 {
 			status = http.StatusBadGateway
 		}
-		c.DataFromReader(status, int64(len(errBody)), "application/json", bytes.NewReader(errBody), nil)
 		c.Header("Content-Type", "application/json")
+		c.DataFromReader(status, int64(len(errBody)), "application/json", bytes.NewReader(errBody), nil)
 		g.recordError(c, ch, resp, start, "upstream_http_"+strconv.Itoa(resp.StatusCode))
 		return
 	}
 
-	// 成功响应
 	c.Header("Content-Type", resp.Header.Get("Content-Type"))
 	c.Status(http.StatusOK)
 	if stream {
-		g.streamCopy(c, ch, resp.Body, start, outUsage)
+		g.streamCopy(c, ch, resp.Body, start, plan.lineConv, sink)
 	} else {
-		g.copyAndCapture(c, ch, resp.Body, start, outUsage)
+		g.copyAndCapture(c, ch, resp.Body, start, plan.bodyConv, sink)
 	}
 }
 
-// copyAndCapture 非流式：整体转发 + 解析 usage + 记账。
-func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, r io.Reader, start time.Time, outUsage func(json.RawMessage)) {
+// copyAndCapture 非流式：整体转发（可转换）+ 解析 usage + 记账。
+func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, r io.Reader, start time.Time, bodyConv func([]byte) ([]byte, error), sink *usageSink) {
 	data, err := io.ReadAll(r)
 	if err != nil {
-		openAIError(c, http.StatusBadGateway, "upstream_error", "failed reading upstream response")
+		apiError(c, http.StatusBadGateway, "upstream_error", "failed reading upstream response")
 		g.recordError(c, ch, nil, start, "read_error")
 		return
 	}
-	// 尝试解析 usage（chat / responses 字段不同）
-	if usageRaw := extractUsage(data); usageRaw != nil {
-		outUsage(usageRaw)
+	if usageRaw := extractUsage(data); usageRaw != nil && sink != nil {
+		sink.push(usageRaw)
 	}
-	_, _ = c.Writer.Write(data)
+	out := data
+	if bodyConv != nil {
+		if converted, cerr := bodyConv(data); cerr == nil {
+			out = converted
+		}
+	}
+	_, _ = c.Writer.Write(out)
 	g.finishUsage(c, ch, start, store.UsageStatusSuccess, "")
 }
 
-// streamCopy 流式：边读上游 SSE 边写客户端，零缓冲转发；扫描 usage 行记账。
-// 客户端断连（ctx cancel）即中止上游读取。
-func (g *Gateway) streamCopy(c *gin.Context, ch *store.Channel, r io.Reader, start time.Time, outUsage func(json.RawMessage)) {
+// streamCopy 流式：边读上游 SSE 边写客户端，零缓冲转发；按 lineConv 转换；扫描 usage 记账。
+func (g *Gateway) streamCopy(c *gin.Context, ch *store.Channel, r io.Reader, start time.Time, lineConv func([]byte) []byte, sink *usageSink) {
 	w := c.Writer
 	flusher, ok := w.(http.Flusher)
 	if !ok {
@@ -158,19 +161,26 @@ func (g *Gateway) streamCopy(c *gin.Context, ch *store.Channel, r io.Reader, sta
 	for {
 		line, err := scanner.Next()
 		if line != nil {
-			if _, werr := w.Write(line); werr != nil {
-				// 客户端断开：取消上游（ctx cancel 由 request ctx 处理）
-				g.recordError(c, ch, nil, start, "client_disconnect")
-				return
+			out := line
+			if lineConv != nil {
+				out = lineConv(line)
 			}
-			flusher.Flush()
-			if usageRaw := scanUsage(line); usageRaw != nil {
-				outUsage(usageRaw)
+			if out != nil {
+				if _, werr := w.Write(out); werr != nil {
+					g.recordError(c, ch, nil, start, "client_disconnect")
+					return
+				}
+				flusher.Flush()
+			}
+			if usageRaw := scanUsage(line); usageRaw != nil && sink != nil {
+				sink.push(usageRaw)
 			}
 		}
 		if err != nil {
 			if err == io.EOF {
 				g.finishUsage(c, ch, start, store.UsageStatusSuccess, "")
+			} else if c.Request.Context().Err() != nil {
+				g.recordError(c, ch, nil, start, "client_disconnect")
 			} else {
 				g.recordError(c, ch, nil, start, "stream_read_error")
 			}
@@ -186,37 +196,34 @@ func (nopFlusher) Flush() {}
 // ---------------------------------------------------------------------------
 // usage 提取
 
-// usageShape 兼容 chat (prompt/completion) 与 responses (input/output) 两种命名。
+// usageShape 兼容 chat (prompt/completion)、responses (input/output)、messages (input/output) 命名。
 type usageShape struct {
-	PromptTokens     int64 `json:"prompt_tokens"`
-	CompletionTokens int64 `json:"completion_tokens"`
-	InputTokens      int64 `json:"input_tokens"`
-	OutputTokens     int64 `json:"output_tokens"`
-	TotalTokens      int64 `json:"total_tokens"`
-	// Claude 缓存口径（M3 接入）
-	CacheReadInputTokens    int64 `json:"cache_read_input_tokens"`
+	PromptTokens             int64 `json:"prompt_tokens"`
+	CompletionTokens         int64 `json:"completion_tokens"`
+	InputTokens              int64 `json:"input_tokens"`
+	OutputTokens             int64 `json:"output_tokens"`
+	TotalTokens              int64 `json:"total_tokens"`
+	CacheReadInputTokens     int64 `json:"cache_read_input_tokens"`
 	CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"`
 }
 
-// extractUsage 从完整响应体提取 usage 子对象。
+// extractUsage 从完整响应体提取 usage 子对象（chat / responses / messages）。
 func extractUsage(data []byte) json.RawMessage {
 	var m map[string]json.RawMessage
 	if json.Unmarshal(data, &m) != nil {
 		return nil
 	}
-	if u, ok := m["usage"]; ok && string(u) != "null" {
+	if u := usageFromMap(m); u != nil {
 		return u
 	}
-	// responses 事件/响应：usage 嵌套在 response 对象内
 	if respRaw, ok := m["response"]; ok {
 		var resp map[string]json.RawMessage
 		if json.Unmarshal(respRaw, &resp) == nil {
-			if u, ok := resp["usage"]; ok && string(u) != "null" {
+			if u := usageFromMap(resp); u != nil {
 				return u
 			}
 		}
 	}
-	// chat 兜底：choices[].message.usage
 	if choices, ok := m["choices"]; ok {
 		var cs []map[string]json.RawMessage
 		if json.Unmarshal(choices, &cs) == nil {
@@ -224,7 +231,7 @@ func extractUsage(data []byte) json.RawMessage {
 				if msgRaw, ok := ch["message"]; ok {
 					var msg map[string]json.RawMessage
 					if json.Unmarshal(msgRaw, &msg) == nil {
-						if u, ok := msg["usage"]; ok && string(u) != "null" {
+						if u := usageFromMap(msg); u != nil {
 							return u
 						}
 					}
@@ -235,7 +242,7 @@ func extractUsage(data []byte) json.RawMessage {
 	return nil
 }
 
-// scanUsage 从 SSE 一行中提取 usage（OpenAI 末块 / responses completed 事件）。
+// scanUsage 从 SSE 一行中提取 usage（OpenAI 末块 / responses completed / messages message_delta 等）。
 func scanUsage(line []byte) json.RawMessage {
 	s := string(line)
 	if !strings.Contains(s, `"usage"`) {
@@ -252,14 +259,29 @@ func scanUsage(line []byte) json.RawMessage {
 	if json.Unmarshal([]byte(s), &m) != nil {
 		return nil
 	}
-	if u, ok := m["usage"]; ok && string(u) != "null" {
+	if u := usageFromMap(m); u != nil {
 		return u
 	}
-	// responses 流式：usage 在 response 对象内（response.completed 事件）
 	if respRaw, ok := m["response"]; ok {
 		var resp map[string]json.RawMessage
 		if json.Unmarshal(respRaw, &resp) == nil {
-			if u, ok := resp["usage"]; ok && string(u) != "null" {
+			if u := usageFromMap(resp); u != nil {
+				return u
+			}
+		}
+	}
+	return nil
+}
+
+// usageFromMap 从 map 顶层或 message 子对象中取 usage。
+func usageFromMap(m map[string]json.RawMessage) json.RawMessage {
+	if u, ok := m["usage"]; ok && string(u) != "null" {
+		return u
+	}
+	if msgRaw, ok := m["message"]; ok {
+		var msg map[string]json.RawMessage
+		if json.Unmarshal(msgRaw, &msg) == nil {
+			if u, ok := msg["usage"]; ok && string(u) != "null" {
 				return u
 			}
 		}
@@ -268,7 +290,6 @@ func scanUsage(line []byte) json.RawMessage {
 }
 
 // sseScanner 按 SSE 行边界读取（兼容 \n 与 \r\n），保留原始行内容。
-// 基于 bufio.Reader：行内可含任意内容，跨 chunk 自动拼接。
 type sseScanner struct {
 	r *bufio.Reader
 }
@@ -289,17 +310,43 @@ func (s *sseScanner) Next() ([]byte, error) {
 // ---------------------------------------------------------------------------
 // 记账
 
-// usageSink 累积流式多次 usage（取最后一次，即最终值）。
+// usageSink 累积多次 usage：合并各事件字段（message_start 给 input，message_delta 给 output）。
 type usageSink struct {
-	last json.RawMessage
+	us usageShape
 }
 
 func (u *usageSink) push(raw json.RawMessage) {
-	if len(raw) > 0 {
-		u.last = raw
+	if len(raw) == 0 {
+		return
+	}
+	var t usageShape
+	if json.Unmarshal(raw, &t) != nil {
+		return
+	}
+	// 零值不覆盖：不同事件携带不同字段
+	if t.PromptTokens > 0 {
+		u.us.PromptTokens = t.PromptTokens
+	}
+	if t.CompletionTokens > 0 {
+		u.us.CompletionTokens = t.CompletionTokens
+	}
+	if t.InputTokens > 0 {
+		u.us.InputTokens = t.InputTokens
+	}
+	if t.OutputTokens > 0 {
+		u.us.OutputTokens = t.OutputTokens
+	}
+	if t.CacheReadInputTokens > 0 {
+		u.us.CacheReadInputTokens = t.CacheReadInputTokens
+	}
+	if t.CacheCreationInputTokens > 0 {
+		u.us.CacheCreationInputTokens = t.CacheCreationInputTokens
 	}
 }
 
+// Shape 返回合并后的用量。
+func (u *usageSink) Shape() usageShape { return u.us }
+
 // finishUsage 落账：计算成本并异步写入。
 func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time, status, errCode string) {
 	uid, _ := c.Get(CtxUserID)
@@ -308,8 +355,8 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time
 
 	var us usageShape
 	if h, ok := c.Get("usage_raw"); ok {
-		if holder, ok := h.(*sinkHolder); ok && holder.sink != nil && len(holder.sink.last) > 0 {
-			_ = json.Unmarshal(holder.sink.last, &us)
+		if holder, ok := h.(*sinkHolder); ok && holder.sink != nil {
+			us = holder.sink.Shape()
 		}
 	}
 
@@ -340,43 +387,57 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time
 		p = "chat"
 	}
 	traceStr, _ := trace.(string)
-	errMsg := errCode
 	latency := int(time.Since(start).Milliseconds())
 
-	// 已写响应头但流中途出错：记 error
 	if status == store.UsageStatusSuccess && c.Writer.Status() >= 400 {
 		status = store.UsageStatusError
 	}
 
+	var errCodePtr *string
+	if errCode != "" {
+		errCodePtr = &errCode
+	}
+
+	var uidVal, kidVal uint64
+	if u, ok := uid.(uint64); ok {
+		uidVal = u
+	}
+	if k, ok := kid.(uint64); ok {
+		kidVal = k
+	}
+	var chID uint64
+	if ch != nil {
+		chID = ch.ID
+	}
+
 	g.rec.Record(&store.UsageLog{
-		RequestID:   fmt.Sprintf("trace-%s", traceStr),
-		TraceID:     traceStr,
-		UserID:      uid.(uint64),
-		KeyID:       kid.(uint64),
-		ChannelID:   ch.ID,
-		ModelID:     modelID,
-		ModelName:   mn,
-		Protocol:    p,
-		InputTokens: in,
-		OutputTokens: out,
-		CacheReadTokens: cacheRead,
+		RequestID:           fmt.Sprintf("trace-%s", traceStr),
+		TraceID:             traceStr,
+		UserID:              uidVal,
+		KeyID:               kidVal,
+		ChannelID:           chID,
+		ModelID:             modelID,
+		ModelName:           mn,
+		Protocol:            p,
+		InputTokens:         in,
+		OutputTokens:        out,
+		CacheReadTokens:     cacheRead,
 		CacheCreationTokens: cacheCreate,
-		InputPrice:  model.InputPrice,
-		OutputPrice: model.OutputPrice,
-		CacheReadPrice: model.CacheReadPrice,
-		Cost:        cost,
-		LatencyMS:   latency,
-		Status:      status,
-		ErrorCode:   &errMsg,
-		CreatedAt:   time.Now().UTC(),
+		InputPrice:          model.InputPrice,
+		OutputPrice:         model.OutputPrice,
+		CacheReadPrice:      model.CacheReadPrice,
+		Cost:                cost,
+		LatencyMS:           latency,
+		Status:              status,
+		ErrorCode:           errCodePtr,
+		CreatedAt:           time.Now().UTC(),
 	})
 }
 
 // recordError 失败请求的记账（不产生扣费，status=error）。
 func (g *Gateway) recordError(c *gin.Context, ch *store.Channel, resp *http.Response, start time.Time, code string) {
-	status := store.UsageStatusError
 	_ = resp
-	g.finishUsage(c, ch, start, status, code)
+	g.finishUsage(c, ch, start, store.UsageStatusError, code)
 }
 
 func now() time.Time { return time.Now() }