fix(codearts): 上游限流改为可声明重试(内核通用能力)+ 会话/连接两处真因

用户连续报「codeart 还是不可用 / empty stream」。把上游原始响应打出来逐条对齐,
三个症状同源,且都不是"模型坏了":

| 场景 | 上游实际返回 |
|---|---|
| 流式、短时间连续请求 | HTTP **200** + 只带 error 的帧 ⇒ 内核判 `empty stream` |
| 非流式、同一条件 | HTTP 400 `TM.00001041: 并发会话数已达上限(3个)` |
| 突发请求 | HTTP 429 `TPM limit … 500000 tokens per minute` |

实测:最后一个请求结束后约 **44~60 秒**释放;同一账号连续请求 3~4 次即占满。
参考实现(示例插件 llm-adapter.ts)的做法是**每 10s 轮询后重试 chat**(最多 30
分钟),所以用户那边"看起来没事"。

## 我们的改法(内核只做通用机制,厂商知识留在插件)
1. **内核新增声明式重试策略** `source.retry`(纯数据:状态码列表 + 响应体子串 +
   间隔 + 次数)。内核不认识任何厂商;命中即在同一网关请求内每 10s 重试一次。
   流式只在**握手阶段**重试(首帧尚未下发,不会重复输出),非流式整体重试。
2. **插件为 codearts 声明**:400/429 + 命中 `TM.00001041` / `TPM limit` /
   `并发会话数已达上限` → 10s × 最多 18 次(≈3 分钟)。
3. **适配器补两个会话细节**:`Connection: close`(keep-alive 空闲连接会被上游
   计入并发会话)与稳定 `Session-Id`(同一账号恒定,避免每次请求新建会话)。
4. 错误文案补可行动提示(并发会话上限 / TPM 限流,约 1 分钟释放)。

## 验证(生产)
- 连续 6 次**流式**:6/6 成功(撞上限时自动等 10s 再试,客户端只感觉慢):
  `+0s / +12s / +34s / +86s / +100s / +144s`
- 连续 5 次**非流式**:5/5 成功(`+0 / +11 / +20 / +28 / +38s`)
- 修复前:第 4 个请求起必然 TM.00001041 / empty stream。

> 注:`TM.00001005 无正常可用套餐` 属于账号权益问题(不重试、如实报出),
> 与限流要分开处理。
This commit is contained in:
root
2026-10-20 10:38:00 +08:00
parent 771b95a133
commit 6454b094bc
3 changed files with 201 additions and 5 deletions

View File

@ -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 {