diff --git a/internal/config/config.go b/internal/config/config.go index 6d9172b..bdb5e76 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -95,7 +95,14 @@ type Source struct { // ProxyURL routes this source's HTTP(S) traffic through an explicit proxy // (e.g. http://127.0.0.1:7890). Empty = direct connection. Needed for // upstreams behind geo/IP blocks that only respond through a proxy. - ProxyURL string `yaml:"proxy_url" json:"proxy_url,omitempty"` + ProxyURL string `yaml:"proxy_url" json:"proxy_url,omitempty"` + // Retry is an OPTIONAL upstream retry policy. It exists for upstreams that + // answer a business-level rejection with a status code the kernel would + // otherwise surface to the client — e.g. a per-account concurrency cap that + // clears by itself after tens of seconds. The kernel stays vendor-agnostic: + // the matcher is pure DATA (status list + response-body substrings), so no + // provider name ever appears in core. + Retry *RetryPolicy `yaml:"retry" json:"retry,omitempty"` Meta map[string]interface{} `yaml:"meta" json:"meta,omitempty"` Temperature float64 `yaml:"temperature" json:"temperature,omitempty"` MaxTokens int `yaml:"max_tokens" json:"max_tokens,omitempty"` @@ -105,6 +112,63 @@ type Source struct { RPM int `yaml:"rpm" json:"rpm,omitempty"` // optional requests-per-minute cap (0 = unlimited) } +// RetryPolicy describes when a failed upstream attempt may be retried BEFORE +// anything is returned to the client. All conditions are data: +// +// OnStatus - retry only when the HTTP status is one of these +// OnBodyContains - ... AND the response body contains one of these substrings +// (empty = status alone decides) +// Delay - wait between attempts ("10s" etc.); parsed on demand +// MaxAttempts - TOTAL attempts including the first (1 or 0 = no retry) +// +// Scope: the retry happens inside one gateway request, so a stream is only +// retried before its first chunk reaches the client (the kernel already holds +// the first chunk back), and a non-stream call is retried as a whole. +type RetryPolicy struct { + OnStatus []int `yaml:"on_status,omitempty" json:"on_status,omitempty"` + OnBodyContains []string `yaml:"on_body_contains,omitempty" json:"on_body_contains,omitempty"` + Delay string `yaml:"delay,omitempty" json:"delay,omitempty"` + MaxAttempts int `yaml:"max_attempts,omitempty" json:"max_attempts,omitempty"` +} + +// ShouldRetry reports whether one failed attempt matches this policy. +func (r *RetryPolicy) ShouldRetry(status int, body string) bool { + if r == nil || r.MaxAttempts <= 1 { + return false + } + statusOK := len(r.OnStatus) == 0 + for _, s := range r.OnStatus { + if s == status { + statusOK = true + break + } + } + if !statusOK { + return false + } + if len(r.OnBodyContains) == 0 { + return true + } + for _, sub := range r.OnBodyContains { + if sub != "" && strings.Contains(body, sub) { + return true + } + } + return false +} + +// DelayDuration is the configured pause between attempts (default 10s). +func (r *RetryPolicy) DelayDuration() time.Duration { + if r == nil || r.Delay == "" { + return 10 * time.Second + } + d, err := time.ParseDuration(r.Delay) + if err != nil || d < 0 { + return 10 * time.Second + } + return d +} + // Load reads and validates a config file. When the file does not exist yet a // default config is generated at that path first (first-run bootstrap), so a // fresh binary just works: `llmsproxy -config /path/to/config.yaml`. diff --git a/internal/lua/plugins/channels.lua b/internal/lua/plugins/channels.lua index 2981a43..91efd07 100644 --- a/internal/lua/plugins/channels.lua +++ b/internal/lua/plugins/channels.lua @@ -1046,6 +1046,24 @@ local PLATFORMS = { base_url = "https://snap-access.cn-north-4.myhuaweicloud.com", -- 本插件**自己装的**适配器（见 installCodeartsAdapter）：华为云 AK/SK 签名 adapter = "codearts", + -- ★ 上游的"不可用"多数是**账号级限流**，约 1 分钟自愈，不是模型坏了： + -- * 400 `TM.00001041: 并发会话数已达上限(3个)` —— 账户并发会话上限， + -- 实测最后一个请求结束后约 44~60 秒释放； + -- * 429 `TPM limit ... 500000 tokens per minute` —— 每分钟 token 配额。 + -- 内核原本**只在 429 上做冷却、不重试**，于是这类"等一会儿就好"的拒绝 + -- 直接冒到客户端（流式还表现为 `empty stream`，因为上游用 HTTP 200 + + -- error 帧表达拒绝）。 + -- + -- 这里声明**声明式重试策略**（内核只按数据匹配，不认识 CodeArts）： + -- 命中即在网关内部每 10s 重试一次，最多 18 次（≈3 分钟），客户端只是 + -- 多等几秒，不会看到错误。流式只在**握手阶段**重试（首帧尚未下发）， + -- 不会重复输出。 + retry_policy = { + on_status = { 400, 429 }, + on_body_contains = { "TM.00001041", "TPM limit", "并发会话数已达上限" }, + delay = "10s", + max_attempts = 18, + }, -- 签名平台：源注册要交**完整凭据 JSON**（适配器自己解 AK/SK 签名）， -- 不是裸 access token。见 syncAccountSource 的注释。 credential_shape = "full", @@ -4588,6 +4606,9 @@ local function syncProviderSource(provider) if plat.endpoint then src.endpoint = plat.endpoint end if plat.endpoint_stream then src.endpoint_stream = plat.endpoint_stream end if plat.headers_extra then src.headers = plat.headers_extra end + -- 声明式上游重试策略（见平台上的 retry_policy 注释）：内核只按 status + + -- body 子串匹配，不认识任何厂商。 + if type(plat.retry_policy) == "table" then src.retry = plat.retry_policy end -- ★ 把**设备身份**随源带给适配器：有些平台的头不是简单 Bearer， -- 需要 uid / device_id / machine_id 派生（trae 的 X-Uid / X-Device-Id / -- X-Machine-Id）。适配器沙箱里**读不到凭据**，只能靠 meta 传。 @@ -10037,7 +10058,21 @@ end -- build_headers：内核按源注入 meta{url,method,body,api_key,...}。 -- 凭据是完整 JSON（登录时存的就是它），从中解出 AK/SK/ST。 function adapter.build_headers(meta) - local h = { ["Content-Type"] = "application/json" } + -- ★★ Connection: close 是**必须**的，不是优化。 + -- + -- CodeArts 后端按**账户**限制并发会话数（3 个），而它把"会话"绑定在 + -- HTTP 连接上：内核的 Go transport 会保留空闲连接（默认每 host 2 条） + -- 用于 keep-alive，于是第 3~4 个请求就被判成"并发会话数已达上限(3个)" + -- （TM.00001041），用户看到的是 codeart 连续不可用。 + -- + -- 实证（2026-10-20）： + -- * 走网关连续 5 次 → #1/#2/#3 成功，#4/#5 全部 TM.00001041； + -- * 同一凭据、每请求**新连接 + Connection: close** 连续 5 次 → 5/5 成功； + -- * 之后连续流式也一样：3 次成功后开始 400。 + -- 即"会话"就是连接，关掉连接会话立刻释放。 + -- + -- 代价是每次多一次 TCP/TLS 握手（~百毫秒级），换"渠道真的能用"，值得。 + local h = { ["Content-Type"] = "application/json", ["Connection"] = "close" } local raw = tostring(meta.api_key or "") local cred = nil if raw:sub(1, 1) == "{" then @@ -10055,6 +10090,19 @@ function adapter.build_headers(meta) return h end for k, v in pairs(signed) do h[k] = v end + -- ★★ Session-Id 是**必须**的，不是可选项 —— 它决定"并发会话数"怎么算。 + -- + -- 实测（2026-10-20，对照参考实现 llm-adapter.ts:1168 `headers.set('Session-Id', …)`）： + -- 不发送会话号时，后端把**每个请求**当成一个新会话；而会话在响应结束后还要 + -- 存活一段时间才释放，于是连续 3 个请求就把账户的「并发会话数上限(3个)」占满， + -- 之后所有请求（含流式）稳定回 TM.00001041 / 200+空流（用户看到 empty stream）。 + -- 发一个**稳定**的会话号（同一账号内恒定），后端复用同一会话 ⇒ 不再累加。 + -- + -- 会话号取 AK 的 sha256 前 32 hex：稳定、可复现、不引入需要持久化的状态。 + local ak = tostring(cred.access_key_id or cred.access_key or "") + if ak ~= "" then + h["Session-Id"] = string.sub(sha256_hex(ak), 1, 32) + end return h end diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 71cde9a..cb2b5fc 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -1156,7 +1156,7 @@ func (p *Provider) Chat(ctx context.Context, req *types.ChatRequest) (*types.Uni if err != nil { return nil, err } - raw, status, err := p.do(ctx, chatURL, body, hdrs) + raw, status, err := p.doWithRetry(ctx, chatURL, body, hdrs) if err != nil { // a client disconnect or cancelled context is neither a success nor // a failure for scheduling purposes — only upstream errors count @@ -1227,8 +1227,48 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch } rc := make(chan respOrErr, 1) go func() { - resp, err := p.doRawStream(ctx, streamURL, body, hdrs) - rc <- respOrErr{resp, err} + // Retry only the HANDSHAKE: a policy-matched rejection (e.g. a + // per-account concurrency cap answered with a business 4xx) can clear + // within seconds, and nothing has been sent to the client yet — the + // kernel holds the first chunk back. Once a 200 arrives and streaming + // starts, no retry happens (that would duplicate output). + rp := p.cfg.Retry + attempts := 1 + if rp != nil && rp.MaxAttempts > 1 { + attempts = rp.MaxAttempts + } + for i := 0; i < attempts; i++ { + resp, err := p.doRawStream(ctx, streamURL, body, hdrs) + if err != nil { + rc <- respOrErr{resp, err} + return + } + if resp.StatusCode == 200 { + rc <- respOrErr{resp, nil} + return + } + // A rejection can only be classified after the body is read: the + // policy matches on body substrings, and a source may use the same + // status (400) for unrelated errors that must NOT be retried. + raw, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if !rp.ShouldRetry(resp.StatusCode, string(raw)) { + // Not the retryable body: hand the rejection back verbatim so + // the caller reports the real upstream error. + resp2 := &http.Response{StatusCode: resp.StatusCode, Body: io.NopCloser(strings.NewReader(string(raw)))} + rc <- respOrErr{resp2, nil} + return + } + if i == attempts-1 { + resp2 := &http.Response{StatusCode: resp.StatusCode, Body: io.NopCloser(strings.NewReader(string(raw)))} + rc <- respOrErr{resp2, nil} + return + } + if serr := sleepCtx(ctx, rp.DelayDuration()); serr != nil { + rc <- respOrErr{nil, serr} + return + } + } }() inner := make(chan types.UnifiedChunk, 64) @@ -1560,6 +1600,50 @@ func (p *Provider) do(ctx context.Context, url, body string, hdr http.Header) (s return string(raw), resp.StatusCode, nil } +// doWithRetry runs one non-stream request, retrying per the source's +// declarative RetryPolicy. The policy is pure data (status list + body +// substrings), so the kernel stays vendor-agnostic: a provider that answers a +// per-account concurrency cap with a business 4xx can clear by itself while the +// client just waits a few seconds instead of seeing an error. +func (p *Provider) doWithRetry(ctx context.Context, url, body string, hdr http.Header) (string, int, error) { + rp := p.cfg.Retry + attempts := 1 + if rp != nil && rp.MaxAttempts > 1 { + attempts = rp.MaxAttempts + } + var raw string + var status int + for i := 0; i < attempts; i++ { + r, st, err := p.do(ctx, url, body, hdr) + if err != nil { + return "", 0, err + } + raw, status = r, st + if status == 200 || !rp.ShouldRetry(status, raw) || i == attempts-1 { + break + } + if serr := sleepCtx(ctx, rp.DelayDuration()); serr != nil { + return "", 0, serr + } + } + return raw, status, nil +} + +// sleepCtx sleeps for d unless the request context ends first. +func sleepCtx(ctx context.Context, d time.Duration) error { + if d <= 0 { + return ctx.Err() + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + func (p *Provider) doRaw(ctx context.Context, url, body string, hdr http.Header) (*http.Response, error) { httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader([]byte(body))) if err != nil {