perf: 目录缓存与流式正文快路径，请求热路径开销减半

按性能分析逐项落地，实测对比（16 并发、mock 上游零延迟、SQLite）：

- 每请求 SQL 10 → 2：模型/渠道目录（模型行、绑定、渠道行）进程内缓存 10s，
  运行期状态（冷却、健康）仍每次实时判定；管理端写操作经中间件、
  健康状态变化经回调立即失效。新增 TestProxyRequestSQLBudget 钉住预算。
- 端到端 p50 1.3–1.6ms → 0.66–0.70ms，RPS 3.5–8.9k → 16–18k。
- 流式正文累积：字符串 += 改 strings.Builder（原为 O(n²)），
  逐行 JSON 提取改字节快路径，并加廉价的子串预过滤、拿到输出 usage 后停累积。
  500 块：949µs/7.9MB 分配 → 83µs/136KB。
- 请求体定点改写：模型映射从整包 unmarshal/marshal 改为顶层字段替换
  （100KB：585µs → 29µs），stream_options 注入同理；requestText 改为
  只在「上游没报用量」时按需提取，亲和键改用原始体前缀。
- 渠道满载背压：新增 Acquire 有界排队（OT_PROXY_CHANNEL_ACQUIRE_WAIT，
  默认 150ms）；排队仍满时返回 503 + Retry-After 而不是 502。
  实测渠道上限 4、16 并发：无排队 81 次 503；排队 150ms 后 400/400 成功。
- 记账按批聚合：同批同用户一次扣款、流水批量插入、日表按 (user,model,日期)
  聚合，替代每行 5 条语句；幂等仍靠 ref_id 唯一索引兜底。
- gorm 开启 PrepareStmt（SQL 解析此前占 CPU 15%）；tiktoken 编码器按模型复用。
- 管理端用量列表不再返回原始请求/响应体（可能几十 MB），
  改为 has_raw 标记 + GET /admin/usage/:id 按需拉取，前端弹窗打开时取。

验证：go vet、go test ./...、go test -race（proxy/channel/usage/tokenizer）、
vue-tsc、eslint、vite build 全绿；本地起 mock 上游实测列表/详情接口与
管理页弹窗按需加载。

server/internal/proxy/gateway.go

@@ -52,12 +52,63 @@ type Gateway struct {
 	firstTokenWait time.Duration
 	// 会话亲和 TTL（0=关闭）：同一会话粘住上次应答的渠道，保住上游提示词缓存。
 	affinityTTL time.Duration
-	aff         *affinityTable
-	tpMu        sync.Mutex
-	tps         map[time.Duration]*http.Transport
+	// 渠道并发满载时每个候选最多排队多久（0=立刻跳过）。
+	channelAcquireWait time.Duration
+	aff                *affinityTable
+	tpMu               sync.Mutex
+	tps                map[time.Duration]*http.Transport
 
 	policyMu sync.Mutex
 	policy   modelPolicy
+
+	// 模型目录缓存：每请求原本要查「是否存在」与「计价」两次同名查询，
+	// 这里按名称缓存 catalogTTL；管理端写操作立即失效。
+	modelMu  sync.Mutex
+	modelCat map[string]modelEntry
+	modelAt  time.Time
+}
+
+// catalogTTL 目录缓存有效期。运行期状态（冷却、健康）不受它影响，
+// 它们每次读取时重新判定；这里只缓存「配置类」数据（模型、渠道、绑定）。
+const catalogTTL = 10 * time.Second
+
+type modelEntry struct {
+	m  store.Model
+	ok bool
+}
+
+// modelByName 取模型（带进程内缓存）；第二个返回值表示模型存在且启用。
+func (g *Gateway) modelByName(name string) (store.Model, bool) {
+	g.modelMu.Lock()
+	if g.modelCat != nil && time.Since(g.modelAt) < catalogTTL {
+		if e, ok := g.modelCat[name]; ok {
+			g.modelMu.Unlock()
+			return e.m, e.ok
+		}
+	}
+	g.modelMu.Unlock()
+
+	var m store.Model
+	err := g.db.Where("name = ? AND enabled = ?", name, true).First(&m).Error
+	e := modelEntry{m: m, ok: err == nil}
+
+	g.modelMu.Lock()
+	if g.modelCat == nil || time.Since(g.modelAt) >= catalogTTL {
+		g.modelCat = map[string]modelEntry{}
+		g.modelAt = time.Now()
+	}
+	g.modelCat[name] = e
+	g.modelMu.Unlock()
+	return e.m, e.ok
+}
+
+// InvalidateCatalog 清空模型与渠道目录缓存（管理端改动、健康状态变化后调用）。
+func (g *Gateway) InvalidateCatalog() {
+	g.modelMu.Lock()
+	g.modelCat = nil
+	g.modelAt = time.Time{}
+	g.modelMu.Unlock()
+	g.ch.InvalidateCatalog()
 }
 
 // modelPolicy 全局模型限制策略（来自系统配置，短时缓存）。
@@ -141,9 +192,7 @@ func (g *Gateway) checkModelExists(c *gin.Context, model string) bool {
 		apiError(c, http.StatusBadRequest, "invalid_request_error", "model is required")
 		return false
 	}
-	var n int64
-	g.db.Model(&store.Model{}).Where("name = ? AND enabled = ?", model, true).Count(&n)
-	if n == 0 {
+	if _, ok := g.modelByName(model); !ok {
 		apiError(c, http.StatusNotFound, "model_not_found",
 			fmt.Sprintf("The model %q does not exist or is not enabled", model))
 		return false
@@ -163,33 +212,36 @@ func contains(list []string, s string) bool {
 // Options 网关配置项。用结构体而不是十个位置参数：里面三个 time.Duration、
 // 两个 int，传错的编译期发现不了。
 type Options struct {
-	UserRPS         int
-	LogRaw          bool
-	GWTimeout       time.Duration // 上游超时兜底（渠道 timeout_ms 优先）
-	MaxBodyMB       int           // /v1 请求体上限（MB）
-	FailoverBudget  time.Duration // 故障转移重试的时间预算
-	StreamHeartbeat time.Duration // 流式空闲心跳间隔
-	FirstTokenWait  time.Duration // 流式放行前等内容的最长时间（0=不限）
-	AffinityTTL     time.Duration // 会话亲和 TTL（0=关闭）
+	UserRPS            int
+	LogRaw             bool
+	GWTimeout          time.Duration // 上游超时兜底（渠道 timeout_ms 优先）
+	MaxBodyMB          int           // /v1 请求体上限（MB）
+	FailoverBudget     time.Duration // 故障转移重试的时间预算
+	StreamHeartbeat    time.Duration // 流式空闲心跳间隔
+	FirstTokenWait     time.Duration // 流式放行前等内容的最长时间（0=不限）
+	AffinityTTL        time.Duration // 会话亲和 TTL（0=关闭）
+	ChannelAcquireWait time.Duration // 渠道满载排队时长（0=立刻跳过）
 }
 
 func NewGateway(db *gorm.DB, enc *crypto.Encryptor, rec *usage.Recorder, lim *ratelimit.Limiter, o Options) *Gateway {
 	return &Gateway{
-		db:              db,
-		ch:              channel.NewService(db, enc),
-		rec:             rec,
-		enc:             enc,
-		lim:             lim,
-		userRPS:         o.UserRPS,
-		logRaw:          o.LogRaw,
-		gwTimeout:       o.GWTimeout,
-		maxBodyMB:       o.MaxBodyMB,
-		failoverBudget:  o.FailoverBudget,
-		streamHeartbeat: o.StreamHeartbeat,
-		firstTokenWait:  o.FirstTokenWait,
-		affinityTTL:     o.AffinityTTL,
-		aff:             newAffinityTable(),
-		tps:             map[time.Duration]*http.Transport{},
+		db:                 db,
+		ch:                 channel.NewService(db, enc),
+		rec:                rec,
+		enc:                enc,
+		lim:                lim,
+		userRPS:            o.UserRPS,
+		logRaw:             o.LogRaw,
+		gwTimeout:          o.GWTimeout,
+		maxBodyMB:          o.MaxBodyMB,
+		failoverBudget:     o.FailoverBudget,
+		streamHeartbeat:    o.StreamHeartbeat,
+		firstTokenWait:     o.FirstTokenWait,
+		affinityTTL:        o.AffinityTTL,
+		channelAcquireWait: o.ChannelAcquireWait,
+		aff:                newAffinityTable(),
+		modelCat:           map[string]modelEntry{},
+		tps:                map[time.Duration]*http.Transport{},
 	}
 }
 
@@ -493,17 +545,15 @@ func prepareUpstream(ch *store.Channel, clientProto string, body []byte, upstrea
 	return plan, nil
 }
 
-// rewriteModel 改写请求体中的 model 字段（三种协议 model 都在顶层）。
+// rewriteModel 改写请求体顶层的 model 字段（模型映射）。
+// 定点替换而不是整包 unmarshal/marshal：后者在 100KB 请求体上实测 ~585µs、
+// 470KB 分配，而这里只有一次扫描 + 拼接。
 func rewriteModel(body []byte, model string) ([]byte, error) {
-	var m map[string]any
-	if err := json.Unmarshal(body, &m); err != nil {
-		return body, nil
+	out, ok := rewriteModelField(body, model)
+	if !ok {
+		return body, nil // 没有顶层 model（非法或空体）：原样转发，交给上游判
 	}
-	if cur, _ := m["model"].(string); cur == model {
-		return body, nil
-	}
-	m["model"] = model
-	return json.Marshal(m)
+	return out, nil
 }
 
 var errNoChannel = errors.New("no available channel")

server/internal/proxy/hold_test.go

@@ -34,6 +34,7 @@ func addChannel(t *testing.T, h *harness, name, baseURL string, weight int) {
 	if err := h.db.Create(&store.ChannelModelBinding{ChannelID: ch.ID, ModelID: m.ID, UpstreamModel: "m1"}).Error; err != nil {
 		t.Fatal(err)
 	}
+	h.gw.InvalidateCatalog() // 管理端写操作会失效目录缓存，测试里直接写库要手动失效
 }
 
 // pushWeightTo 把指定名字渠道的权重改大，让加权随机几乎必然从它开始。
@@ -42,6 +43,7 @@ func pushWeightTo(t *testing.T, h *harness, name string, weight int) {
 	if err := h.db.Model(&store.Channel{}).Where("name = ?", name).Update("weight", weight).Error; err != nil {
 		t.Fatal(err)
 	}
+	h.gw.InvalidateCatalog()
 }
 
 // 流式容错：渠道1 在首个内容事件之前回错误事件（200+流式形式的错误），应换渠道2

server/internal/proxy/jsonfast.go

@@ -0,0 +1,155 @@
+package proxy
+
+import (
+	"bytes"
+	"encoding/json"
+)
+
+// 请求体上的定点改写：只动需要改的那一段，不做整包 unmarshal → map → marshal。
+// 整包往返在 agent 场景（上下文几十到几百 KB）实测约 0.5–0.6ms 且分配翻倍，
+// 而这里只做一次 O(n) 扫描。
+
+// rewriteModelField 改写请求体顶层 model 字段，其余字节原样保留。
+// 嵌套出现的 "model"（比如消息内容里引用的 JSON）不会被误改：只认顶层键。
+// 找不到顶层 model 或请求体不是对象时返回 (body, false)。
+func rewriteModelField(body []byte, model string) ([]byte, bool) {
+	span, ok := topLevelValueSpan(body, "model")
+	if !ok {
+		return body, false
+	}
+	quoted, err := json.Marshal(model)
+	if err != nil {
+		return body, false
+	}
+	out := make([]byte, 0, len(body)-(span[1]-span[0])+len(quoted))
+	out = append(out, body[:span[0]]...)
+	out = append(out, quoted...)
+	out = append(out, body[span[1]:]...)
+	return out, true
+}
+
+// injectStreamOptions 在请求体顶层追加 stream_options，让流式末块带上 usage。
+// 只在没有该键时插入；body 必须是 JSON 对象（以 '}' 结尾，允许尾部空白）。
+func injectStreamOptions(body []byte, options string) ([]byte, bool) {
+	trimmed := bytes.TrimRight(body, " \t\r\n")
+	if len(trimmed) < 2 || trimmed[0] != '{' || trimmed[len(trimmed)-1] != '}' {
+		return body, false
+	}
+	if bytes.Contains(trimmed, []byte(`"stream_options"`)) {
+		return body, false
+	}
+	end := len(trimmed) - 1
+	out := make([]byte, 0, len(body)+len(options)+16)
+	out = append(out, trimmed[:end]...)
+	// 空对象 {} 不能加逗号
+	if last := bytes.TrimRight(trimmed[:end], " \t\r\n"); len(last) > 1 {
+		out = append(out, ',')
+	}
+	out = append(out, `"stream_options":`...)
+	out = append(out, options...)
+	out = append(out, '}')
+	return out, true
+}
+
+// topLevelValueSpan 找顶层键 key 的值在 body 中的字节区间 [start, end)。
+// 只遍历顶层（depth==1）的键，跳过字符串字面量与转义。
+func topLevelValueSpan(body []byte, key string) ([2]int, bool) {
+	var zero [2]int
+	trimmed := bytes.TrimLeft(body, " \t\r\n")
+	if len(trimmed) == 0 || trimmed[0] != '{' {
+		return zero, false
+	}
+	// 重新对齐到 trimmed 起点后的偏移
+	base := len(body) - len(trimmed)
+	keyPat := []byte(`"` + key + `"`)
+	i := base + 1
+	depth := 1
+	for i < len(body) {
+		switch body[i] {
+		case '{', '[':
+			depth++
+			i++
+		case '}', ']':
+			depth--
+			if depth == 0 {
+				return zero, false
+			}
+			i++
+		case '"':
+			start := i
+			i = skipJSONString(body, i)
+			// 顶层字符串：可能是"键"，也可能是"值"
+			if depth == 1 && bytes.Equal(body[start:i], keyPat) {
+				j := i
+				for j < len(body) && (body[j] == ' ' || body[j] == '\t' || body[j] == '\r' || body[j] == '\n') {
+					j++
+				}
+				if j < len(body) && body[j] == ':' {
+					j++
+					for j < len(body) && (body[j] == ' ' || body[j] == '\t' || body[j] == '\r' || body[j] == '\n') {
+						j++
+					}
+					vs := j
+					ve, ok := jsonValueEnd(body, vs)
+					if !ok {
+						return zero, false
+					}
+					return [2]int{vs, ve}, true
+				}
+			}
+		default:
+			i++
+		}
+	}
+	return zero, false
+}
+
+// skipJSONString 返回从 body[i]（'"'）开始的下一个未转义引号之后的位置。
+func skipJSONString(body []byte, i int) int {
+	i++ // 跳过开引号
+	for i < len(body) {
+		switch body[i] {
+		case '\\':
+			i += 2
+		case '"':
+			return i + 1
+		default:
+			i++
+		}
+	}
+	return i
+}
+
+// jsonValueEnd 返回从 i 开始的一个 JSON 值的结束位置（不含尾部空白）。
+func jsonValueEnd(body []byte, i int) (int, bool) {
+	if i >= len(body) {
+		return 0, false
+	}
+	switch body[i] {
+	case '"':
+		return skipJSONString(body, i), true
+	case '{', '[':
+		depth := 0
+		for j := i; j < len(body); j++ {
+			switch body[j] {
+			case '"':
+				j = skipJSONString(body, j) - 1
+			case '{', '[':
+				depth++
+			case '}', ']':
+				depth--
+				if depth == 0 {
+					return j + 1, true
+				}
+			}
+		}
+		return 0, false
+	default:
+		j := i
+		for j < len(body) && body[j] != ',' && body[j] != '}' && body[j] != ']' &&
+			body[j] != ' ' && body[j] != '\t' && body[j] != '\r' && body[j] != '\n' {
+			j++
+		}
+		return j, j > i
+	}
+}

server/internal/proxy/jsonfast_test.go

@@ -0,0 +1,100 @@
+package proxy
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func TestRewriteModelFieldTopLevelOnly(t *testing.T) {
+	in := `{"model":"a","messages":[{"role":"user","content":"{\"model\":\"nested\"}"}],"max_tokens":5}`
+	out, ok := rewriteModelField([]byte(in), "b")
+	if !ok {
+		t.Fatal("应改写顶层 model")
+	}
+	var m map[string]any
+	if err := json.Unmarshal(out, &m); err != nil {
+		t.Fatalf("输出必须是合法 JSON: %v\n%s", err, out)
+	}
+	if m["model"] != "b" {
+		t.Fatalf("顶层 model 未改写：%v", m["model"])
+	}
+	// 嵌套的 model 引用必须原样保留
+	if !strings.Contains(string(out), `\"model\":\"nested\"`) {
+		t.Fatalf("嵌套内容被改动了：%s", out)
+	}
+	if m["max_tokens"].(float64) != 5 {
+		t.Fatalf("其它字段丢失：%v", out)
+	}
+}
+
+func TestRewriteModelFieldKeepsByteShape(t *testing.T) {
+	in := "{\n  \"model\" : \"old\",\n  \"stream\": true\n}"
+	out, ok := rewriteModelField([]byte(in), "new")
+	if !ok {
+		t.Fatal("应改写")
+	}
+	want := "{\n  \"model\" : \"new\",\n  \"stream\": true\n}"
+	if string(out) != want {
+		t.Fatalf("应只替换值本身，实际：%q", out)
+	}
+}
+
+func TestRewriteModelFieldMissing(t *testing.T) {
+	if _, ok := rewriteModelField([]byte(`{"stream":true}`), "x"); ok {
+		t.Fatal("没有 model 时不应改写")
+	}
+	if _, ok := rewriteModelField([]byte(`not json`), "x"); ok {
+		t.Fatal("非 JSON 不应改写")
+	}
+}
+
+func TestInjectStreamOptions(t *testing.T) {
+	cases := []struct{ in, want string }{
+		{`{"model":"m"}`, `{"model":"m","stream_options":{"include_usage":true}}`},
+		{"{\n  \"model\": \"m\"\n}", "{\n  \"model\": \"m\"\n,\"stream_options\":{\"include_usage\":true}}"},
+		{`{}`, `{"stream_options":{"include_usage":true}}`}, // 空对象也能安全追加
+	}
+	for _, c := range cases {
+		out, ok := injectStreamOptions([]byte(c.in), `{"include_usage":true}`)
+		if !ok {
+			t.Fatalf("应注入：%s", c.in)
+		}
+		if string(out) != c.want {
+			t.Fatalf("注入结果 = %q，期望 %q", out, c.want)
+		}
+		var m map[string]any
+		if err := json.Unmarshal(out, &m); err != nil {
+			t.Fatalf("注入后不是合法 JSON: %v (%s)", err, out)
+		}
+		if _, ok := m["stream_options"]; !ok {
+			t.Fatalf("缺少 stream_options: %s", out)
+		}
+	}
+	// 已有 stream_options 时不重复注入
+	if _, ok := injectStreamOptions([]byte(`{"stream_options":{"include_usage":true}}`), `{}`); ok {
+		t.Fatal("已有该键时不应再注入")
+	}
+}
+
+// 快路径必须与 JSON 解析结果一致（含转义、非正文事件、三协议形态）。
+func TestFastContentTextMatchesJSON(t *testing.T) {
+	lines := []string{
+		`data: {"choices":[{"delta":{"content":"hello"}}]}`,
+		`data: {"choices":[{"delta":{"content":"中文内容"}}]}`,
+		`data: {"choices":[{"delta":{"content":"带\"引号\"和\\反斜杠"}}]}`,
+		`data: {"choices":[{"delta":{"role":"assistant"}}]}`,
+		`data: {"choices":[{"delta":{},"finish_reason":"stop"}]}`,
+		`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"块文本"}}`,
+		`data: {"type":"response.output_text.delta","delta":"响应增量"}`,
+		`data: {"type":"message_start","message":{"usage":{}}}`,
+		`data: [DONE]`,
+	}
+	for _, line := range lines {
+		got := fastContentText([]byte(line))
+		want := sseContentText([]byte(line))
+		if got != want {
+			t.Fatalf("不一致\n行: %s\n快路径: %q\nJSON: %q", line, got, want)
+		}
+	}
+}

server/internal/proxy/passthrough.go

@@ -43,8 +43,10 @@ func parseBody(c *gin.Context) (*bodyReq, []byte, error) {
 		return nil, nil, err
 	}
 	c.Request.Body = io.NopCloser(bytes.NewReader(body))
-	// 中断请求输入文本（token 估算用，仅存文本不立即分词）
-	c.Set("est_input_text", requestText(body))
+	// 只留存原始请求体：输入文本提取（requestText）要整包 JSON 解析，
+	// 而它只在「上游没报用量」时才用得到（实测 100KB 请求体 ~506µs）。
+	// 亲和键与估算都改成用到时再提取。
+	c.Set("req_body", body)
 	br := &bodyReq{}
 	_ = json.Unmarshal(body, br) // 解析失败按空处理，直通仍可转发
 	return br, body, nil
@@ -162,6 +164,47 @@ func sseContentText(line []byte) string {
 	return ""
 }
 
+// fastContentText 从一条 SSE 行里取出增量正文，走字节扫描而不是整行 JSON 解析。
+// 逐条 JSON 反序列化到 map 大约 1–2µs、二十多次分配，一份长回答几千条事件就是
+// 毫秒级纯开销；这里只在三种固定键位上定位字符串值，遇到转义再回退到 JSON 解析。
+//
+// 覆盖三协议的增量形态：
+//   - chat：{"choices":[{"delta":{"content":"…"}}]}
+//   - messages：{"type":"content_block_delta","delta":{"text":"…"}}
+//   - responses：{"type":"response.output_text.delta","delta":"…"}
+func fastContentText(line []byte) string {
+	for _, key := range []string{`"content":"`, `"text":"`, `"delta":"`} {
+		idx := bytes.Index(line, []byte(key))
+		if idx < 0 {
+			continue
+		}
+		start := idx + len(key) // 指向值的第一个字符（引号已含在 key 里）
+		end := start
+		escaped := false
+		for end < len(line) {
+			switch line[end] {
+			case '\\':
+				escaped = true
+				end += 2
+				continue
+			case '"':
+				if escaped {
+					// 有转义：这一段交给 JSON 解析，保证还原正确
+					var out string
+					if json.Unmarshal(line[idx+len(key)-1:end+1], &out) == nil {
+						return out
+					}
+					return ""
+				}
+				return string(line[start:end])
+			}
+			end++
+		}
+		return "" // 行被截断：留给下一行
+	}
+	return ""
+}
+
 // retryAfter 解析上游 Retry-After（delta-秒或 HTTP 日期），用于按对端意愿延长渠道冷却。
 // 解析不出或为非正返回 0，交给 channel.Service 用默认时长（并夹到上限）。
 func retryAfter(h http.Header) time.Duration {
@@ -236,6 +279,9 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto
 	// 最后真正试过的渠道：全部失败时把错误归到它头上，否则 usage_logs 里
 	// channel_id=0，渠道维度的统计与排障就看不到这类失败。
 	var lastCh *store.Channel
+	// saturated：有候选因并发满载被跳过。一次都没打到上游就全被跳过时，
+	// 那不是「上游坏了」（502），而是「我方满了」（503 + Retry-After）。
+	saturated := false
 	startAll := time.Now()
 	// 从选中的渠道开始遍历，到末尾后再从头遍历到选中渠道之前
 	for offset := 0; offset < len(cands); offset++ {
@@ -262,9 +308,10 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto
 			lastStatus, lastBody = http.StatusInternalServerError, []byte("conversion error: "+err.Error())
 			continue
 		}
-		release, ok := g.ch.TryAcquire(ch)
+		release, ok := g.ch.Acquire(c.Request.Context(), ch, g.channelAcquireWait)
 		if !ok {
-			continue // 渠道满载，溢出到下一个
+			saturated = true // 满载：稍等仍拿不到槽，溢出到下一个
+			continue
 		}
 		// 失败/成功计数由 proxyOne 就地报：只有它读得到上游响应头（429 的 Retry-After）
 		lastCh = ch
@@ -291,6 +338,19 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto
 		}
 		lastStatus, lastBody = st, b
 	}
+	// 全部候选都没打出去、且都是被并发满载挡住的：503 让客户端稍后重试，
+	// 而不是 502（那意味着上游故障，客户端通常会直接放弃）。
+	if saturated && lastCh == nil {
+		c.Header("Retry-After", "1")
+		msg := "all upstream channels are at capacity, retry shortly"
+		if sh != nil && sh.headSent() {
+			_ = sh.writeDirect(streamErrorEvent(clientProto, http.StatusServiceUnavailable, msg))
+		} else {
+			apiError(c, http.StatusServiceUnavailable, "channel_saturated", msg)
+		}
+		g.recordError(c, nil, nil, now(), "channel_saturated")
+		return
+	}
 	// 全部候选重试性失败
 	if sh != nil && sh.headSent() {
 		_ = sh.writeDirect(streamErrorEvent(clientProto, lastStatus, string(lastBody)))
@@ -313,9 +373,15 @@ func (g *Gateway) affinityKeyFor(c *gin.Context) string {
 	}
 	mn, _ := c.Get("model_name")
 	model, _ := mn.(string)
-	text, _ := c.Get("est_input_text")
-	firstText, _ := text.(string)
-	return affinityKey(kID, model, firstText)
+	// 亲和键取原始请求体的前缀：每轮重发全量上下文时开头是稳定的（模型、系统提示、
+	// 最早的几条消息），足够区分会话；不必为了它整包解析 JSON。
+	var prefix string
+	if v, ok := c.Get("req_body"); ok {
+		if raw, ok2 := v.([]byte); ok2 {
+			prefix = string(raw) // affinityKey 内部只取前 512 字节
+		}
+	}
+	return affinityKey(kID, model, prefix)
 }
 
 // streamErrorEvent 把错误构造成客户端协议的 SSE 错误事件（响应头已写出时的唯一交付方式）。
@@ -349,12 +415,9 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan
 	upBody := plan.body
 	// 直通 chat 流式：注入 stream_options.include_usage，保证末块带 usage（OpenAI 行为）
 	if stream && plan.path == "/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}
-			if b, err := json.Marshal(m); err == nil {
-				upBody = b
-			}
+		// 定点追加，避免整包 unmarshal/marshal（大请求体上是几百微秒 + 一倍分配）
+		if b, ok := injectStreamOptions(upBody, `{"include_usage":true}`); ok {
+			upBody = b
 		}
 	}
 
@@ -546,7 +609,7 @@ func (g *Gateway) streamHoldCopy(c *gin.Context, sh *streamHold, ch *store.Chann
 					return true, false, false, 0, nil
 				}
 				if trySink != nil {
-					trySink.outputText += sseContentText(out)
+					trySink.addContent(out)
 				}
 			}
 			if !releaseSeen && sseEventHasContent(out, clientProto) {
@@ -773,10 +836,37 @@ func extractServedModel(data []byte) string {
 // outputText 累积已转发内容文本，用于流式中断时按 tiktoken 估算输出 token。
 type usageSink struct {
 	us          usageShape
-	outputText  string
 	servedModel string // 上游实际应答的模型名（模型映射/聚合商场景下与请求的不同）
+
+	// text 累积已转发的正文，仅用于「上游没报输出用量」时按 tiktoken 估算。
+	// 用 Builder 而非字符串拼接：后者每块整体拷贝，长回答是 O(n²)
+	// （实测 10KB/500 块要 949µs、7.9MB 分配）。
+	text strings.Builder
+	// sawOutput：上游已经报过输出 token，估算不再需要，后面不再累积正文。
+	sawOutput bool
 }
 
+// textCap 正文累积上限：超过后不再累积（估算本来只是近似，不值得为它吃满内存）。
+const textCap = 512 << 10
+
+// addContent 从一条已转发的 SSE 行里取出正文并累积。
+// 先做廉价的子串预过滤，只有可能携带正文的行才走 JSON 解析——一份流里
+// role/usage/finish 之类的元事件占比不低，逐行 JSON 解析是纯浪费。
+func (u *usageSink) addContent(line []byte) {
+	if u.sawOutput || u.text.Len() >= textCap {
+		return
+	}
+	if !bytes.Contains(line, []byte(`"content"`)) && !bytes.Contains(line, []byte(`"delta"`)) && !bytes.Contains(line, []byte(`"text"`)) {
+		return
+	}
+	if s := fastContentText(line); s != "" {
+		u.text.WriteString(s)
+	}
+}
+
+// textOf 已累积的正文。
+func (u *usageSink) textOf() string { return u.text.String() }
+
 func (u *usageSink) push(raw json.RawMessage) {
 	if len(raw) == 0 {
 		return
@@ -793,6 +883,11 @@ func (u *usageSink) push(raw json.RawMessage) {
 		u.us = t
 		return
 	}
+	// 一旦上游报过输出 token，正文估算就没有意义了（也就不用再累积）
+	if t.CompletionTokens > 0 || t.OutputTokens > 0 {
+		u.sawOutput = true
+		u.text.Reset()
+	}
 	// 零值不覆盖：不同事件携带不同字段
 	if t.PromptTokens > 0 {
 		u.us.PromptTokens = t.PromptTokens
@@ -850,7 +945,10 @@ func (u *usageSink) merge(o *usageSink) {
 	if o.us.CacheCreationInputTokens > 0 {
 		u.us.CacheCreationInputTokens = o.us.CacheCreationInputTokens
 	}
-	u.outputText += o.outputText
+	if !u.sawOutput && !o.sawOutput {
+		u.text.WriteString(o.text.String())
+	}
+	u.sawOutput = u.sawOutput || o.sawOutput
 	if o.servedModel != "" && u.servedModel == "" {
 		u.servedModel = o.servedModel
 	}
@@ -907,21 +1005,23 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time
 	// 非流式上游必返回 usage，此处 in/out 非 0 不受影响。
 	if status == store.UsageStatusCanceled || status == store.UsageStatusSuccess {
 		if in == 0 && cacheRead == 0 && cacheCreate == 0 {
-			if est, ok := c.Get("est_input_text"); ok {
-				if v, ok2 := est.(string); ok2 && v != "" {
-					in = int64(tokenizer.Count(v, mn))
+			if v, ok := c.Get("req_body"); ok {
+				if raw, ok2 := v.([]byte); ok2 && len(raw) > 0 {
+					if text := requestText(raw); text != "" {
+						in = int64(tokenizer.Count(text, mn))
+					}
 				}
 			}
 		}
-		if out == 0 && sink != nil && sink.outputText != "" {
-			out = int64(tokenizer.Count(sink.outputText, mn))
+		if out == 0 && sink != nil && sink.textOf() != "" {
+			out = int64(tokenizer.Count(sink.textOf(), mn))
 		}
 	}
 
 	var model store.Model
 	var cost float64
 	var modelID uint64
-	_ = g.db.Where("name = ?", mn).First(&model).Error
+	model, _ = g.modelByName(mn)
 	if model.ID > 0 {
 		modelID = model.ID
 		// 计价口径：in=非缓存输入、cacheRead=缓存读、cacheCreate=缓存写（Anthropic 语义，

server/internal/proxy/sqlbudget_test.go

@@ -0,0 +1,80 @@
+package proxy
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/openteam/server/internal/channel"
+	"gorm.io/gorm"
+	gormlogger "gorm.io/gorm/logger"
+)
+
+// sqlTracer 记录一次请求路径上打过的 SQL。
+type sqlTracer struct {
+	gormlogger.Interface
+	mu   sync.Mutex
+	list []string
+}
+
+func (s *sqlTracer) Trace(_ context.Context, _ time.Time, fc func() (string, int64), _ error) {
+	sql, _ := fc()
+	s.mu.Lock()
+	s.list = append(s.list, sql)
+	s.mu.Unlock()
+}
+
+func (s *sqlTracer) take() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := s.list
+	s.list = nil
+	return out
+}
+
+// 每请求 SQL 预算：热路径上的模型/渠道目录必须走进程内缓存。
+// 原本一次非流式请求要打 10 条（含同一模型查三次），缓存后只剩鉴权两条。
+// 这个上限是防回归的闸门：往请求路径上新增未缓存的查询会立刻让它变红。
+func TestProxyRequestSQLBudget(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"id":"x","model":"m1","choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":5}}`)
+	}))
+	defer srv.Close()
+
+	h := newHarness(t, srv.URL, 0, true, nil)
+	tr := &sqlTracer{Interface: gormlogger.Discard}
+	traced := h.db.Session(&gorm.Session{Logger: tr})
+	h.gw.db = traced
+	h.gw.ch = channel.NewService(traced, h.gw.enc)
+
+	body := `{"model":"m1","messages":[{"role":"user","content":"hi"}]}`
+	// 第一次：冷缓存 + last_used_at 首次写入，不计入预算
+	if w := h.run(t, nil, body); w.Code != http.StatusOK {
+		t.Fatalf("code=%d body=%s", w.Code, w.Body.String())
+	}
+	tr.take()
+
+	// 第二次：模型/渠道目录与系统配置都应命中缓存
+	if w := h.run(t, nil, body); w.Code != http.StatusOK {
+		t.Fatalf("code=%d body=%s", w.Code, w.Body.String())
+	}
+	sqls := tr.take()
+	const budget = 3 // api_keys + users（+ 偶发 last_used_at）
+	if len(sqls) > budget {
+		t.Fatalf("热路径 SQL 条数 = %d，超过预算 %d：\n%s", len(sqls), budget, joinSQL(sqls))
+	}
+	t.Logf("第二条请求 SQL = %d 条：\n%s", len(sqls), joinSQL(sqls))
+}
+
+func joinSQL(sqls []string) string {
+	out := ""
+	for i, q := range sqls {
+		out += "  " + string(rune('1'+i)) + ". " + q + "\n"
+	}
+	return out
+}