diff --git a/internal/gateway/chat.go b/internal/gateway/chat.go index 429ea59..489fc68 100644 --- a/internal/gateway/chat.go +++ b/internal/gateway/chat.go @@ -29,6 +29,9 @@ type chatRequest struct { ToolChoice interface{} `json:"tool_choice,omitempty"` DisableThinking bool `json:"disable_thinking"` ExtraBody map[string]interface{} `json:"extra_body,omitempty"` + + // PromptCacheKey 是 OpenAI 原生的缓存键；部分客户端用它携带会话标识。 + PromptCacheKey string `json:"prompt_cache_key,omitempty"` } // ChatCompletion is the non-streaming OpenAI response object. @@ -283,6 +286,8 @@ func (g *Gateway) handleChat(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error()) return } + // 客户端自带的会话标识（若有）比推导出来的更准，见 clientSessionFromRequest。 + clientSession := clientSessionFromRequest(r, req.PromptCacheKey) if len(req.Messages) == 0 { writeError(w, http.StatusBadRequest, "invalid_request", "messages is required") return @@ -313,6 +318,7 @@ func (g *Gateway) handleChat(w http.ResponseWriter, r *http.Request) { ToolChoice: req.ToolChoice, DisableThinking: req.DisableThinking, ExtraBody: req.ExtraBody, + ClientSession: clientSession, } rec := &Req{ Key: keyID(reqKey(ctx)), @@ -368,6 +374,7 @@ func (g *Gateway) handleChat(w http.ResponseWriter, r *http.Request) { ToolChoice: req.ToolChoice, DisableThinking: req.DisableThinking, ExtraBody: req.ExtraBody, + ClientSession: clientSession, } rec := &Req{ Key: keyID(reqKey(ctx)), @@ -534,6 +541,9 @@ var overflowMarkers = []*regexp.Regexp{ } // overflowCanonical 命中 pi 的 /context[_ ]length[_ ]exceeded/i。 +// +// 注意：这个前缀只是必要条件，不是充分条件——pi 还会先用 +// NON_OVERFLOW_PATTERNS 排除整条消息，见 overflowClientMessage。 const overflowCanonical = "context_length_exceeded" // looksLikeOverflow 判断错误文本是否属于上下文超窗。 @@ -552,14 +562,59 @@ func looksLikeOverflow(s string) bool { // Per-tier one-line reasons are kept (quota/cooling skips carry no error body // and are the actionable part); each is capped so HTML dumps can't leak. // -// 超窗会被归一化成 overflowCanonical 前缀，见 overflowMarkers 的说明。 +// 超窗会被归一化成「干净」的 overflowCanonical 消息，见 overflowClientMessage。 func clientUpstreamErr(err error) string { log.Printf("[gateway] upstream failure surfaced to client: %v", err) - msg := upstreamErrSummary(err) - if looksLikeOverflow(err.Error()) || looksLikeOverflow(msg) { - return fmt.Sprintf("%s: %s", overflowCanonical, msg) + if looksLikeOverflow(err.Error()) { + return overflowClientMessage(err) } - return msg + return upstreamErrSummary(err) +} + +// overflowHint 是给客户端的超窗说明，刻意只用 pi 认得的措辞。 +const overflowHint = "context window is full; reduce the length of the messages" + +// clientSessionFromRequest 取客户端自带的会话标识，拿不到时返回 ""。 +// +// 通用 OpenAI 客户端默认不带会话 id，但 pi 支持：只要 provider 的 compat 里打开 +// sendSessionAffinityHeaders，pi 就会把**平台真实的会话 id**（uuidv7，整个会话 +// 恒定）放到 x-session-affinity / x-client-request-id 上（sessionAffinityFormat +// 为 openrouter 时是 x-session-id，为 openai 时是 session_id）。 +// 有了它就不必再从首条 user 消息推导会话指纹（那只在客户端不发会话 id 时才作为 +// 退路，且历史压缩后会漂移）。prompt_cache_key 是 OpenAI 原生的缓存键， +// 客户端若带也一并当会话用。 +// +// 刻意不采纳 x-client-request-id：名字含 request，部分客户端每请求都换， +// 拿它当会话会让上游缓存永不命中。pi 总会同时发 x-session-affinity，够用。 +func clientSessionFromRequest(r *http.Request, bodyKey string) string { + for _, h := range []string{"X-Session-Affinity", "X-Session-Id", "Session-Id"} { + if v := strings.TrimSpace(r.Header.Get(h)); v != "" { + return v + } + } + return strings.TrimSpace(bodyKey) +} + +// overflowClientMessage 为超窗失败生成「干净」的客户端消息。 +// +// 为什么不能沿用 upstreamErrSummary：pi 的 isContextOverflow 先查 +// NON_OVERFLOW_PATTERNS（/rate limit/、/too many requests/、Bedrock 前缀）， +// 一旦命中就直接判为「非超窗」——**哪怕消息里已经有 context_length_exceeded**， +// pi 也不会压缩重试。而 AUTO 链的失败消息天生是多 tier 原因的拼接： +// 超窗 tier（gozen 400 maximum context length）常与配额/限流 tier +// （429 token plan exhausted、cooling、no free slot）同时出现。 +// 把明细原样带出去，等于让一条限流 tier 的措辞反过来封杀超窗识别。 +// 所以这里只保留「超窗」措辞 + 是哪个源超的窗，其余一律不带。 +func overflowClientMessage(err error) string { + var ce *scheduler.ChainErr + if errors.As(err, &ce) { + for _, t := range ce.Tiers { + if looksLikeOverflow(t.Err.Error()) { + return fmt.Sprintf("%s: %s (%s/%s)", overflowCanonical, overflowHint, t.Source, t.Model) + } + } + } + return fmt.Sprintf("%s: %s", overflowCanonical, overflowHint) } // upstreamErrSummary 把上游失败压成一行短消息。 diff --git a/internal/gateway/client_session_test.go b/internal/gateway/client_session_test.go new file mode 100644 index 0000000..3868ce5 --- /dev/null +++ b/internal/gateway/client_session_test.go @@ -0,0 +1,47 @@ +package gateway + +import ( + "net/http" + "testing" +) + +// 客户端自带的会话标识必须被采纳：它比「首条 user 消息指纹」更准 +// （历史压缩后指纹会漂移），也是唯一能天然按会话隔离的来源。 +func TestClientSessionFromRequestHeaders(t *testing.T) { + cases := []struct { + name string + headers map[string]string + bodyKey string + want string + }{ + {"pi 默认格式 x-session-affinity", map[string]string{"X-Session-Affinity": "ses-1"}, "", "ses-1"}, + {"openrouter 格式 x-session-id", map[string]string{"X-Session-Id": "ses-2"}, "", "ses-2"}, + {"openai 格式 session_id", map[string]string{"Session-Id": "ses-3"}, "", "ses-3"}, + {"body prompt_cache_key 兜底", nil, "cache-4", "cache-4"}, + {"头优先于 body", map[string]string{"X-Session-Affinity": "ses-5"}, "cache-x", "ses-5"}, + {"空白不算", map[string]string{"X-Session-Affinity": " "}, "", ""}, + {"都没有则为空（适配器自行推导）", nil, "", ""}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r, _ := http.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + for k, v := range tc.headers { + r.Header.Set(k, v) + } + if got := clientSessionFromRequest(r, tc.bodyKey); got != tc.want { + t.Errorf("clientSessionFromRequest = %q, want %q", got, tc.want) + } + }) + } +} + +// 刻意不采纳 x-client-request-id：名字含 request，部分客户端每请求都换， +// 拿它当会话会让上游前缀缓存永不命中。 +func TestClientRequestIDIsNotUsedAsSession(t *testing.T) { + r, _ := http.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + r.Header.Set("X-Client-Request-Id", "per-request-uuid") + if got := clientSessionFromRequest(r, ""); got != "" { + t.Errorf("x-client-request-id 不应被当作会话（否则缓存永不命中），got %q", got) + } +} diff --git a/internal/gateway/overflow_err_test.go b/internal/gateway/overflow_err_test.go index 28110fb..632719a 100644 --- a/internal/gateway/overflow_err_test.go +++ b/internal/gateway/overflow_err_test.go @@ -94,3 +94,74 @@ func TestOverflowPhraseSurvivesTruncationWidth(t *testing.T) { t.Fatalf("归一化标记丢失: %q", got) } } + +// piNonOverflowRe 是 @earendil-works/pi-ai 的 NON_OVERFLOW_PATTERNS。 +// pi 的 isContextOverflow 先查这一组，命中则直接判为「非超窗」—— +// 即使 OVERFLOW_PATTERNS 也命中，也不会压缩重试。 +var piNonOverflowRe = []*regexp.Regexp{ + regexp.MustCompile(`^(Throttling error|Service unavailable):`), + regexp.MustCompile(`(?i)rate limit`), + regexp.MustCompile(`(?i)too many requests`), +} + +// piWouldCompact 复刻 isContextOverflow 的 Case 1 判定顺序： +// 先排除 NON_OVERFLOW，再匹配 OVERFLOW。 +func piWouldCompact(errorMessage string) bool { + for _, p := range piNonOverflowRe { + if p.MatchString(errorMessage) { + return false + } + } + return piOverflowRe.MatchString(errorMessage) +} + +// issue 回归：AUTO 链失败消息天然把「超窗 tier」和「配额/限流 tier」拼在一起， +// 超窗标记后面只要出现 rate limit / too many requests 的字样， +// pi 就会把整条消息判为「非超窗」而拒绝压缩重试。 +func TestOverflowMessageSurvivesRateLimitedSiblingTier(t *testing.T) { + cases := []struct { + name string + sibling string + }{ + {"rate limit 措辞", "api error 429: rate limit exceeded"}, + {"too many requests 措辞", "429 Too Many Requests"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := &scheduler.ChainErr{Tiers: []scheduler.TierError{ + {Tier: 1, Source: "sensenova", Model: "glm-5.2", Err: errors.New(tc.sibling)}, + {Tier: 3, Source: "gozen", Model: "deepseek-v4.1-flash", + Err: errors.New("api error 400: This model's maximum context length is 1048576 tokens")}, + }} + + got := clientUpstreamErr(err) + if !piWouldCompact(got) { + t.Fatalf("同链限流 tier 封杀了超窗识别，pi 不会压缩重试: %q", got) + } + // 干净：不得把限流 tier 的措辞带出去。 + for _, p := range piNonOverflowRe { + if p.MatchString(got) { + t.Fatalf("超窗消息仍带限流措辞 %v: %q", p, got) + } + } + // 但仍要能看出是哪个源超的窗。 + if !strings.Contains(got, "gozen/deepseek-v4.1-flash") { + t.Errorf("超窗源信息丢失: %q", got) + } + }) + } +} + +// 直连超窗同样要干净可用。 +func TestDirectOverflowMessageIsClean(t *testing.T) { + raw := "api error 400: Error from provider (Console Go): This model's maximum context length is 1048576 tokens. " + + "However, you requested 1440050 tokens. Please reduce the length of the messages or completion." + got := clientUpstreamErr(errors.New(raw)) + if !piWouldCompact(got) { + t.Fatalf("直连超窗未被 pi 识别: %q", got) + } + if !strings.Contains(got, overflowCanonical) { + t.Fatalf("缺少归一化标记: %q", got) + } +} diff --git a/internal/lua/adapters/opencodego.lua b/internal/lua/adapters/opencodego.lua index e936f69..c2a9038 100644 --- a/internal/lua/adapters/opencodego.lua +++ b/internal/lua/adapters/opencodego.lua @@ -62,6 +62,33 @@ local function conversation_fingerprint(body) return "" end +-- session_seed 决定上游会话号（x-opencode-session 的种子）。 +-- +-- 优先级： +-- 1) 客户端自带的会话标识（meta.client_session）——pi 等在开启 +-- compat.sendSessionAffinityHeaders 后会发 x-session-affinity， +-- 这是平台真实的会话 id，整个会话恒定且天然按会话隔离。 +-- 2) 退路：历史里**第一条 user 消息**做会话指纹。 +-- 3) 再退：只按源固定（连 user 消息都没有时）。 +-- +-- 为什么需要 2/3：通用 OpenAI 客户端默认**根本不发**会话标识 —— 实测抓包 +-- （tcpdump 抓 127.0.0.1:8081 真实 agent 请求）确认 body 里没有 +-- user / session_id / conversation_id / metadata，请求头也只有 +-- X-Stainless-*（OpenAI JS SDK）与 User-Agent。opencode 原生客户端那套 +-- x-opencode-session 是它自己的概念，通用客户端无从转发。 +-- +-- 为什么必须稳定：上游前缀缓存是**会话级**的。会话号每请求一变， +-- 缓存永不命中（实测：固定会号第 2 次命中 5888，每请求换会号则恒为 0）。 +local function session_seed(meta) + local src = (meta.source and meta.source.name) or "" + local base = "session|llmsproxy|" .. src + local cs = meta.client_session + if type(cs) == "string" and cs ~= "" then + return base .. "|client|" .. cs + end + return base .. "|" .. conversation_fingerprint(meta.body) +end + function adapter.build_headers(meta) local ts = tostring(meta.timestamp or "") local src = (meta.source and meta.source.name) or "" @@ -76,7 +103,7 @@ function adapter.build_headers(meta) -- 每请求换 session -> 永远 0 命中 -- 原先用 meta.timestamp 派生，等于每请求都是新会话，缓存永远无效， -- 上游也无法做会话亲和路由。 - ["x-opencode-session"] = rand_id("ses_", "session|llmsproxy|" .. src .. "|" .. conversation_fingerprint(meta.body)), + ["x-opencode-session"] = rand_id("ses_", session_seed(meta)), -- request id 仍每请求唯一（它只是请求标识，不参与缓存键） ["x-opencode-request"] = rand_id("msg_", "request|" .. ts .. "|" .. tostring(meta.body or "")), } diff --git a/internal/lua/adapters/opencodezen.lua b/internal/lua/adapters/opencodezen.lua index 4a65149..de0a540 100644 --- a/internal/lua/adapters/opencodezen.lua +++ b/internal/lua/adapters/opencodezen.lua @@ -63,6 +63,33 @@ local function conversation_fingerprint(body) return "" end +-- session_seed 决定上游会话号（x-opencode-session 的种子）。 +-- +-- 优先级： +-- 1) 客户端自带的会话标识（meta.client_session）——pi 等在开启 +-- compat.sendSessionAffinityHeaders 后会发 x-session-affinity， +-- 这是平台真实的会话 id，整个会话恒定且天然按会话隔离。 +-- 2) 退路：历史里**第一条 user 消息**做会话指纹。 +-- 3) 再退：只按源固定（连 user 消息都没有时）。 +-- +-- 为什么需要 2/3：通用 OpenAI 客户端默认**根本不发**会话标识 —— 实测抓包 +-- （tcpdump 抓 127.0.0.1:8081 真实 agent 请求）确认 body 里没有 +-- user / session_id / conversation_id / metadata，请求头也只有 +-- X-Stainless-*（OpenAI JS SDK）与 User-Agent。opencode 原生客户端那套 +-- x-opencode-session 是它自己的概念，通用客户端无从转发。 +-- +-- 为什么必须稳定：上游前缀缓存是**会话级**的。会话号每请求一变， +-- 缓存永不命中（实测：固定会号第 2 次命中 5888，每请求换会号则恒为 0）。 +local function session_seed(meta) + local src = (meta.source and meta.source.name) or "" + local base = "session|llmsproxy|" .. src + local cs = meta.client_session + if type(cs) == "string" and cs ~= "" then + return base .. "|client|" .. cs + end + return base .. "|" .. conversation_fingerprint(meta.body) +end + function adapter.build_headers(meta) local ts = tostring(meta.timestamp or "") local src = (meta.source and meta.source.name) or "" @@ -77,7 +104,7 @@ function adapter.build_headers(meta) -- 每请求换 session -> 永远 0 命中 -- 原先用 meta.timestamp 派生，等于每请求都是新会话，缓存永远无效， -- 上游也无法做会话亲和路由。 - ["x-opencode-session"] = rand_id("ses_", "session|llmsproxy|" .. src .. "|" .. conversation_fingerprint(meta.body)), + ["x-opencode-session"] = rand_id("ses_", session_seed(meta)), -- request id 仍每请求唯一（它只是请求标识，不参与缓存键） ["x-opencode-request"] = rand_id("msg_", "request|" .. ts .. "|" .. tostring(meta.body or "")), } diff --git a/internal/lua/toolcall_preservation_test.go b/internal/lua/toolcall_preservation_test.go index 6cf5cf2..74fb40d 100644 --- a/internal/lua/toolcall_preservation_test.go +++ b/internal/lua/toolcall_preservation_test.go @@ -301,3 +301,53 @@ func TestOpenCodeSessionIsStableForCache(t *testing.T) { vm.Stop() } } + +// 客户端自带的会话标识（pi 的 x-session-affinity）必须优先于会话指纹： +// 它是平台真实会话 id，历史压缩后也不会漂移，且天然按会话隔离。 +func TestOpenCodePrefersClientSessionID(t *testing.T) { + for _, name := range []string{"opencodego", "opencodezen"} { + vm := NewVM(freshAdapterDir(t)) + if err := vm.Start(); err != nil { + t.Fatal(err) + } + meta := func(body, cs string) map[string]interface{} { + return map[string]interface{}{ + "timestamp": 1000, + "body": body, + "client_session": cs, + "source": map[string]interface{}{"name": "somesource", "meta": nil}, + } + } + + // 同一条会话：历史在长，但客户端会话 id 不变。 + a, err := vm.BuildHeaders(name, meta(`{"messages":[{"role":"user","content":"fix the bug"}]}`, "cs-aaa")) + if err != nil { + t.Fatal(err) + } + b, err := vm.BuildHeaders(name, meta(`{"messages":[{"role":"user","content":"a totally different opener"}]}`, "cs-aaa")) + if err != nil { + t.Fatal(err) + } + if a["x-opencode-session"] != b["x-opencode-session"] { + t.Errorf("%s: 客户端会话 id 相同就必须同会话（内容无关）: %q vs %q", + name, a["x-opencode-session"], b["x-opencode-session"]) + } + + // 不同会话。 + c, _ := vm.BuildHeaders(name, meta(`{"messages":[{"role":"user","content":"fix the bug"}]}`, "cs-bbb")) + if a["x-opencode-session"] == c["x-opencode-session"] { + t.Errorf("%s: 不同客户端会话必须不同会号", name) + } + + // 客户端没给会话 id 时，回落到指纹（同内容 -> 同会号）。 + d, _ := vm.BuildHeaders(name, meta(`{"messages":[{"role":"user","content":"fix the bug"}]}`, "")) + e, _ := vm.BuildHeaders(name, meta(`{"messages":[{"role":"user","content":"fix the bug"}]}`, "")) + if d["x-opencode-session"] != e["x-opencode-session"] { + t.Errorf("%s: 无客户端会号时应回落到稳定指纹", name) + } + if d["x-opencode-session"] == a["x-opencode-session"] { + t.Errorf("%s: 客户端会号与指纹两条路必须产出不同会号", name) + } + vm.Stop() + } +} diff --git a/internal/provider/provider.go b/internal/provider/provider.go index c942199..cd9f068 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -707,7 +707,7 @@ func (p *Provider) probeModels(ctx context.Context) (bool, string) { if err != nil { return false, "" } - if hdrs, herr := p.buildHeaders("{}", u); herr == nil { + if hdrs, herr := p.buildHeaders("{}", u, ""); herr == nil { req.Header = hdrs } resp, err := p.client.Do(req) @@ -750,7 +750,7 @@ func (p *Provider) probeChat(ctx context.Context) (bool, string) { body, err := json.Marshal(probe) if err == nil { var hdr http.Header - if hdrs, herr := p.buildHeaders(string(body), p.URL()); herr == nil { + if hdrs, herr := p.buildHeaders(string(body), p.URL(), ""); herr == nil { hdr = hdrs } raw, status, derr := p.do(ctx, p.URL(), string(body), hdr) @@ -1031,13 +1031,15 @@ func (p *Provider) Throttle(ctx context.Context) error { // ---- request construction ---- -func (p *Provider) buildHeaders(body, url string) (http.Header, error) { +func (p *Provider) buildHeaders(body, url, clientSession string) (http.Header, error) { meta := map[string]interface{}{ "url": url, "method": http.MethodPost, "body": body, "api_key": p.cfg.APIKey, "timestamp": types.Now(), + // client_session 是客户端自带的会话标识（请求头），为空时适配器自行推导。 + "client_session": clientSession, "source": map[string]interface{}{ "name": p.cfg.Name, "meta": p.cfg.Meta, @@ -1082,7 +1084,7 @@ func (p *Provider) Chat(ctx context.Context, req *types.ChatRequest) (*types.Uni if err != nil { return nil, err } - hdrs, err := p.buildHeaders(body, p.URL()) + hdrs, err := p.buildHeaders(body, p.URL(), req.ClientSession) if err != nil { return nil, err } @@ -1142,7 +1144,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch p.Release() return nil, err } - hdrs, err := p.buildHeaders(body, p.URL()) + hdrs, err := p.buildHeaders(body, p.URL(), req.ClientSession) if err != nil { p.Release() return nil, err @@ -1411,7 +1413,7 @@ func (p *Provider) Image(ctx context.Context, req *types.ImageGenRequest) (*type // fall back to passthrough adapter (openai-style) transformed = string(b) } - hdrs, err := p.buildHeaders(transformed, p.ImageURL()) + hdrs, err := p.buildHeaders(transformed, p.ImageURL(), "") if err != nil { return nil, err } diff --git a/internal/types/types.go b/internal/types/types.go index 6ff4178..94b94c8 100644 --- a/internal/types/types.go +++ b/internal/types/types.go @@ -29,6 +29,10 @@ type ChatRequest struct { ToolChoice interface{} `json:"tool_choice,omitempty"` DisableThinking bool `json:"disable_thinking"` ExtraBody map[string]interface{} `json:"-"` + + // ClientSession 是客户端自带的会话标识（请求头 x-session-affinity 等）。 + // 不参与序列化：它只用于让适配器生成稳定的上游会话，不应泄漏给上游 body。 + ClientSession string `json:"-"` } func (r *ChatRequest) MarshalJSON() ([]byte, error) {