Merge branch 'feat/m43-logupstreamadopt-adopt-reviewed-comm' into dev
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

This commit is contained in:
AskaEth committed 2026-10-26 14:02:22 +08:00
commit abbd79a3b1
17 files changed
+624 -184

No files matched your search

+11 -18
View File
@@ -14,6 +14,7 @@ import (
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// maxResponseBytes caps how much of an /embeddings response body is read, so an
@@ -21,8 +22,9 @@ import (
// for a normal batch fall far below this; the cap is a guard, not a limit.
const maxResponseBytes = 16 << 20
// maxErrorBodyBytes caps how much of a non-OK response body is read for the
// error message.
// maxErrorBodyBytes 限制错误路径**读多少字节**(只用来数长度、取摘要、抽
// code/type):正文一个字节都不进 error,因此它不再决定「错误里带多少原文」。
// 非 2xx 是攻击面而不是数据面,上界刻意小。
const maxErrorBodyBytes = 4096
// HTTP client tuning values, matching the recommended pool in
@@ -234,8 +236,13 @@ func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector,
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
msg := readErrorBody(resp.Body)
return nil, fmt.Errorf("embedding: status %d: %s", resp.StatusCode, msg)
// 这一处是全片最敏感的一条:/embeddings 的 **input 就是她的记忆正文**,
// 而兼容层常把 input 原样回显在错误体里(`Invalid value: '<正文>'`)。
// 因此正文只用来数长度、取摘要、抽 code/type,一个字节都不进 error ——
// 口径由 trace.UpstreamError(M40 共享 helper)统一决定。
body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes))
return nil, fmt.Errorf("embedding: %w", trace.UpstreamError(
trace.HostOf(e.cfg.APIURL), resp.StatusCode, resp.Header.Get("Content-Type"), body))
}
data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
@@ -300,20 +307,6 @@ func (e *APIEmbedder) newEmbeddingRequest(ctx context.Context, texts []string) (
return req, nil
}
// readErrorBody drains a small prefix of a non-OK response body into a message,
// capped at maxErrorBodyBytes so an error response cannot exhaust memory.
func readErrorBody(r io.Reader) string {
raw, err := io.ReadAll(io.LimitReader(r, maxErrorBodyBytes))
if err != nil {
return err.Error()
}
msg := strings.TrimSpace(string(raw))
if msg == "" {
return "bad status"
}
return msg
}
// defaultHTTPClient builds the HTTP client configuration described in
// DevDocs/02-development-constraints.md §2.3: tuned idle/TLS timeouts plus a
// total request timeout. A fresh client is returned per call (never a package
+36
View File
@@ -135,6 +135,42 @@ func TestAPIEmbedder_Non200Errors(t *testing.T) {
}
}
// TestAPIEmbedder_Non200DropsEchoedInput 覆盖 M41 里最敏感的一处:/embeddings 的
// **input 就是她的记忆正文**,而兼容层常把 input 原样回显在错误体里。旧实现读
// 4096 字节响应体、整个拼进 error,而 error 会被 retrieve_fallback 等位置打日志。
//
// 断言的是「文本里搜不到正文」,不是「error 非空」—— 后者在旧实现下同样成立,
// 打不红。
func TestAPIEmbedder_Non200DropsEchoedInput(t *testing.T) {
t.Parallel()
const memo = "SENTINEL-MEMORY-INPUT-4a7d"
rt := roundTripFunc(func(r *http.Request) (*http.Response, error) {
body := `{"error":{"code":"invalid_request_error","param":"input",` +
`"message":"Invalid value: '` + memo + `'"}}`
return &http.Response{
StatusCode: http.StatusBadRequest,
Body: io.NopCloser(strings.NewReader(body)),
Header: http.Header{"Content-Type": []string{"application/json"}},
}, nil
})
emb := NewAPI(
config.EmbeddingConfig{APIURL: "https://embed.example.com/v1/embeddings"},
WithRoundTripper(rt),
)
_, err := emb.Embed(context.Background(), []string{memo})
if err == nil {
t.Fatal("expected error on 400, got nil")
}
if strings.Contains(err.Error(), memo) {
t.Errorf("error text echoes the memory text: %v", err)
}
if !strings.Contains(err.Error(), "400") {
t.Errorf("err = %v, want the status kept for triage", err)
}
}
func TestAPIEmbedder_MismatchedCountErrors(t *testing.T) {
t.Parallel()
+4 -4
View File
@@ -5,7 +5,6 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"strings"
@@ -314,7 +313,7 @@ func (a *AnthropicChatAdapter) streamOnce(ctx context.Context, req Request, onEv
if resp.StatusCode != http.StatusOK {
return Response{}, a.upstreamError(resp)
}
return scanAnthropicSSE(resp.Body, onEvent)
return scanAnthropicSSE(upstreamLabel(a.name, a.baseURL), resp.Body, onEvent)
}
// resolveModel 返回实际生效的模型名。
@@ -404,9 +403,10 @@ func (a *AnthropicChatAdapter) doRequest(httpReq *http.Request, stream bool) (*h
return resp, nil
}
// upstreamError 把一次非 2xx 响应交给包内唯一的上游错误收口(见 upstream.go):
// 只留 status / code / type 与体积摘要,响应体正文与 error.message 不出镜。
func (a *AnthropicChatAdapter) upstreamError(resp *http.Response) error {
snippet, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
return fmt.Errorf("model: %w: status %d: %s", ErrUpstream, resp.StatusCode, strings.TrimSpace(string(snippet)))
return providerUpstreamError(a.name, a.baseURL, resp)
}
// record 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
+7 -5
View File
@@ -61,8 +61,8 @@ type streamBlock struct {
// tool_call_delta 事件转发(片段由 tool_call_start/tool_call_end 包起来),
// 并按块累积,以便块结束时重新拼装成结构化的 Args map。
// 流必须以 message_stop 事件终止,
// 否则返回 ErrStream。
func scanAnthropicSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) {
// 否则返回 ErrStream。upstream 只用于给流内 error 事件标注「打到哪」。
func scanAnthropicSSE(upstream string, r io.Reader, onEvent func(StreamEvent)) (Response, error) {
scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes)
agg := Response{}
@@ -84,7 +84,7 @@ func scanAnthropicSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error)
if err := json.Unmarshal([]byte(data), &evt); err != nil {
return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err)
}
if err := applyAnthropicEvent(&agg, blocks, &order, &text, evt, onEvent); err != nil {
if err := applyAnthropicEvent(&agg, blocks, &order, &text, evt, onEvent, upstream); err != nil {
return Response{}, err
}
if evt.Type == "message_stop" {
@@ -101,7 +101,8 @@ func scanAnthropicSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error)
return agg, nil
}
// applyAnthropicEvent 把单个流事件折叠进正在运行的聚合状态。
// applyAnthropicEvent 把单个流事件折叠进正在运行的聚合状态。upstream 只用于给
// error 事件标注「打到哪」,不参与任何成功路径。
func applyAnthropicEvent(
agg *Response,
blocks map[int]*streamBlock,
@@ -109,6 +110,7 @@ func applyAnthropicEvent(
text *strings.Builder,
evt anthropicStreamEvent,
onEvent func(StreamEvent),
upstream string,
) error {
switch evt.Type {
case "message_start":
@@ -213,7 +215,7 @@ func applyAnthropicEvent(
}
case "error":
if evt.Error != nil {
return fmt.Errorf("model: %w: anthropic stream error: %s: %s", ErrUpstream, evt.Error.Type, evt.Error.Message)
return streamEventError(upstream, evt.Error)
}
}
return nil
+35
View File
@@ -281,6 +281,41 @@ func TestAnthropicChat_Chat_UpstreamError(t *testing.T) {
}
}
// TestAnthropicChat_ChatStream_ErrorEventDropsMessage 覆盖 M41 的流内错误:
// HTTP 是 200,失败藏在 SSE 的 error 事件里,而 evt.Error.Message 是**无上限**
// 的上游自由文本(旧实现整段拼进 error)。上游会把请求内容回显在这条 message
// 里,因此它一个字都不能进 error。
//
// 保留的诊断面是 evt.Type(协议枚举):它是灰度那条「这是什么失败」的线索。
func TestAnthropicChat_ChatStream_ErrorEventDropsMessage(t *testing.T) {
t.Parallel()
const echoed = "SENTINEL-ANTHROPIC-STREAM-5b1c"
sse := strings.Join([]string{
`data: {"type":"message_start","message":{"model":"claude-test","usage":{"input_tokens":5}}}`,
``,
`data: {"type":"error","error":{"type":"invalid_request_error","message":"prompt is too long: ` + echoed + `"}}`,
``,
}, "\n")
srv := newAnthropicServer(t, http.StatusOK, "text/event-stream", sse, nil)
ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"})
_, err := ad.ChatStream(context.Background(),
Request{Messages: []Message{TextMessage(RoleUser, echoed)}, Model: "claude-test"},
func(StreamEvent) {})
if err == nil {
t.Fatal("want error, got nil")
}
if !errors.Is(err, ErrUpstream) {
t.Fatalf("err = %v, want ErrUpstream", err)
}
if strings.Contains(err.Error(), echoed) {
t.Errorf("error text carries the stream error message: %v", err)
}
if !strings.Contains(err.Error(), "invalid_request_error") {
t.Errorf("err = %v, want the event type kept for triage", err)
}
}
func TestAnthropicChat_Chat_NoBaseURL(t *testing.T) {
t.Parallel()
ad := NewAnthropicChat(config.LLMConfig{AnthropicModel: "m"})
+3 -5
View File
@@ -5,7 +5,6 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"strings"
@@ -24,8 +23,6 @@ const (
sseBufInit = 64 * 1024
// maxSSELineBytes 限制单条 SSE 事件行的长度。
maxSSELineBytes = 1024 * 1024
// errBodyCap 限制为取出消息而读取的错误响应体大小。
errBodyCap = 512
// defaultHTTPTimeout 在 Options.Timeout 为 0 时约束单次提供商调用。
defaultHTTPTimeout = 180 * time.Second
)
@@ -234,9 +231,10 @@ func (a *OpenAIChatAdapter) doRequest(httpReq *http.Request, stream bool) (*http
return resp, nil
}
// upstreamError 把一次非 2xx 响应交给包内唯一的上游错误收口(见 upstream.go):
// 只留 status / code / type 与体积摘要,响应体正文与 error.message 不出镜。
func (a *OpenAIChatAdapter) upstreamError(resp *http.Response) error {
snippet, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
return fmt.Errorf("model: %w: status %d: %s", ErrUpstream, resp.StatusCode, strings.TrimSpace(string(snippet)))
return providerUpstreamError(a.name, a.baseURL, resp)
}
// record 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
+34
View File
@@ -12,6 +12,7 @@ import (
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
const okCompletion = `{
@@ -244,6 +245,39 @@ func TestOpenAIChat_Chat_UpstreamError(t *testing.T) {
}
}
// TestOpenAIChat_Chat_UpstreamErrorDropsPromptEcho 覆盖 M41 的一家 provider:
// 4xx 的错误体里回显了她的提问,**提问不能出现在 error 文本里**。
//
// 为什么断言的是「文本里搜不到」而不是「error 非空」:旧实现把响应体截断
// 512 字节后拼进 error,error 又被 chat run / loop step 打日志 —— 截断只是把
// 「整段泄漏」变成「前 512 字节泄漏」。只有「搜不到原文」这个断言能把实现
// 变异回旧写法时打红。
func TestOpenAIChat_Chat_UpstreamErrorDropsPromptEcho(t *testing.T) {
t.Parallel()
const echoed = "SENTINEL-OPENAI-PROMPT-9f3a"
body := `{"error":{"code":"invalid_request_error","type":"invalid_request_error",` +
`"param":"messages","message":"Invalid value for 'messages[0].content': '` + echoed + `'"}}`
srv := newCompletionServer(t, http.StatusUnprocessableEntity, "application/json", body, nil)
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "m"})
_, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, echoed)}}, Options{})
if err == nil {
t.Fatal("want error, got nil")
}
if !errors.Is(err, ErrUpstream) || !errors.Is(err, trace.ErrUpstream) {
t.Fatalf("err = %v, want both ErrUpstream and trace.ErrUpstream", err)
}
if strings.Contains(err.Error(), echoed) {
t.Errorf("error text echoes the prompt: %v", err)
}
if !strings.Contains(err.Error(), "422") {
t.Errorf("err = %v, want the status kept for triage", err)
}
if !strings.Contains(err.Error(), "invalid_request_error") {
t.Errorf("err = %v, want the error.code kept for triage", err)
}
}
func TestOpenAIChat_Chat_NoBaseURL(t *testing.T) {
t.Parallel()
ad := NewOpenAIChat(config.LLMConfig{})
+92
View File
@@ -0,0 +1,92 @@
package model
import (
"fmt"
"io"
"net/http"
"strings"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// 本文件是 model 包**唯一**的「上游错误体 → error」收口。两个 provider 适配
// (OpenAI 兼容 /chat/completions 与 Anthropic /v1/messages,含流内 error 事件)
// 都经这里进 error,绝不各自截一段响应体再拼进去 —— 那样写的人早晚会漏掉一家,
// 而漏掉的方向永远是「她的提问进日志」。
//
// 出镜口径由 internal/runtime/trace.UpstreamError 决定(M40 的共享 helper,契约见
// 其文件头):只留 upstream / status / contentType / bodyLen / body sha256 前缀 /
// error.code / error.type,**error.message 默认带不进**。这里不重新实现任何清洗
// —— 三家各写一份清洗就是三处漂移的起点。
// errBodyCap 限制错误路径**读多少字节**(只用来数长度、取摘要、抽 code/type),
// 不再决定「错误里带多少正文」:正文一个字节都不进 error。上界刻意小 ——
// 错误体是攻击面,不是数据面。
const errBodyCap = 512
// maxErrorTokenBytes 是 error.code / error.type 允许出镜的长度上限(见
// safeErrorToken)。
const maxErrorTokenBytes = 64
// upstreamLabel 报「这次调用实际打到哪」:优先主机(自建网关 / 代理下才看得出
// 真实去向),取不到主机时退回 provider 名(代码里写死的短名)。两种形态都不含
// 路径、查询串与凭据 —— 契约在 trace.HostOf。
func upstreamLabel(name, baseURL string) string {
if host := trace.HostOf(baseURL); host != "" {
return host
}
return name
}
// providerUpstreamError 把一次非 2xx 响应收敛成**可安全进日志**的错误。
//
// 为什么不再带正文:上游把用户的提问原样回显在错误里是真实写法 ——
// `{"error":{"code":"invalid_request_error","message":"Invalid value: '<她的原话>'"}}`,
// 而这类 error 会被 `chat run` / `loop step` 打日志。截断到 512 字节只是把
// 「整段泄漏」变成「前 512 字节泄漏」,不是修复。
//
// 双 %w:既保持 errors.Is(err, ErrUpstream)(降级链按它分类,主 → 备 → 本地),
// 也让 errors.Is(err, trace.ErrUpstream) 成立(调用方按上游失败分类)。
func providerUpstreamError(name, baseURL string, resp *http.Response) error {
body, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
info := trace.UpstreamError(upstreamLabel(name, baseURL), resp.StatusCode,
resp.Header.Get("Content-Type"), body)
return fmt.Errorf("model: %w: %w", ErrUpstream, info)
}
// streamEventError 处理 /v1/messages 的**流内** error 事件:HTTP 已经是 200,
// 失败藏在 SSE 里,因此没有状态码。
//
// evt.Message 是上游自由文本,Anthropic 会用它回显请求内容(例如「prompt is too
// long: <片段>」),所以**整段丢弃**(原本是无上限拼进 error)。只保留 evt.Type
// (协议枚举,如 overloaded_error)且必须像令牌才出镜:网关把散文塞进 type 时它
// 是空串 —— 宁可少一条分类提示,不让自由文本借道 type 出去。
func streamEventError(upstream string, evt *anthropicStreamError) error {
info, ok := trace.UpstreamError(upstream, 0, "text/event-stream", nil).(*trace.UpstreamErrorInfo)
if !ok {
// helper 的构造是固定的,走不到这里;真走到也只少一条诊断,不破安全口径。
return fmt.Errorf("model: %w: anthropic stream error event", ErrUpstream)
}
info.Type = safeErrorToken(evt.Type)
return fmt.Errorf("model: %w: %w", ErrUpstream, info)
}
// safeErrorToken 只放行「像协议令牌」的短串(字母、数字与 `_ . -`),其余一律
// 丢弃。error.code / error.type 在主流上游是枚举,但它们是**自由文本面**:网关、
// 代理与兼容层可以往里面塞任何东西。丢弃它的代价只是少一条分类提示,保留一个
// 不受控字段的代价是她的话进日志。
func safeErrorToken(s string) string {
s = strings.TrimSpace(s)
if s == "" || len(s) > maxErrorTokenBytes {
return ""
}
for _, r := range s {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
case r == '_' || r == '.' || r == '-':
default:
return ""
}
}
return s
}
@@ -11,6 +11,7 @@ import (
"strings"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// visionMaxTokens 限制 vision/OCR 响应的长度。
@@ -170,8 +171,14 @@ func (a *OpenAICompatibleAdapter) ask(ctx context.Context, model string, img Ima
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
snippet, _ := io.ReadAll(io.LimitReader(resp.Body, maxFetchedBody))
return "", fmt.Errorf("vision: %w: status %d: %s", ErrUpstream, resp.StatusCode, strings.TrimSpace(string(snippet)))
// 错误路径**不再整段读**:这里原来沿用取图的 maxFetchedBody(50 MiB),
// 等于「最多 50 MiB 的错误体进内存、再进 error」。错误体是攻击面,因此
// 只读 maxErrorBodyBytes,并且一个字节都不进 error —— 只留 status /
// contentType / bodyLen / sha256 与 code/type(trace.UpstreamError)。
body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes))
upstream := trace.UpstreamError(trace.HostOf(a.cfg.BaseURL), resp.StatusCode,
resp.Header.Get("Content-Type"), body)
return "", fmt.Errorf("vision: %w: %w", ErrUpstream, upstream)
}
return a.decode(resp.Body)
}
@@ -3,9 +3,11 @@ package vision
import (
"context"
"encoding/json"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
@@ -51,3 +53,62 @@ func TestOpenAICompatibleAdapter_NotConfigured(t *testing.T) {
t.Fatal("want error for empty base url, got nil")
}
}
// stubTransport 让测试不经网络直接给出响应。
type stubTransport func(*http.Request) (*http.Response, error)
func (f stubTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
// countingBody 记录被读走的字节数,用来断言错误路径**有界读**。
type countingBody struct {
r io.Reader
n *int64
}
func (b *countingBody) Read(p []byte) (int, error) {
n, err := b.r.Read(p)
*b.n += int64(n)
return n, err
}
func (b *countingBody) Close() error { return nil }
// TestOpenAICompatibleAdapter_ErrorBodyKeptOutOfError 覆盖 M41 的视觉一处:
// 旧实现沿用取图上限(50 MiB)读错误体,再把它拼进 error —— 既是日志面
// (整个 50 MiB 的错误体都能进去),也是 DoS 面(一次超大错误响应就吃掉内存)。
//
// 两个断言各能单独打红:① 正文出现在 error 里 ⇒ 又回到「拼原文」;
// ② 错误路径读走超过 maxErrorBodyBytes ⇒ 又回到按取图上限整段读。
// 测试体用 8 MiB(远大于 4 KiB、远小于 50 MiB):足以证伪无界读,又不必在
// 测试里分配 50 MiB。
func TestOpenAICompatibleAdapter_ErrorBodyKeptOutOfError(t *testing.T) {
t.Parallel()
const sentinel = "SENTINEL-VISION-ERROR-BODY-7d2a"
var read int64
payload := sentinel + strings.Repeat("x", 8<<20)
ad := newAdapter("https://vision.example.com")
ad.client = &http.Client{Transport: stubTransport(func(*http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusInternalServerError,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: &countingBody{r: strings.NewReader(payload), n: &read},
}, nil
})}
_, err := ad.Describe(context.Background(), ImageRef{Source: SourceBase64, Data: "aGVsbG8="}, "")
if err == nil {
t.Fatal("want error, got nil")
}
if !errors.Is(err, ErrUpstream) {
t.Fatalf("err = %v, want ErrUpstream", err)
}
if strings.Contains(err.Error(), sentinel) {
t.Errorf("error text carries the response body: %v", err)
}
if read > maxErrorBodyBytes {
t.Errorf("error path read %d bytes of the error body, want <= %d", read, maxErrorBodyBytes)
}
if !strings.Contains(err.Error(), "500") {
t.Errorf("err = %v, want the status kept for triage", err)
}
}
@@ -18,6 +18,11 @@ import (
const (
maxImageBytes = 50 << 20
maxFetchedBody = 50 << 20
// maxErrorBodyBytes 是非 2xx 响应的读取上限。它与上面两个是**不同口径**:
// 那两个是数据面(图像字节)的预算,这一条是攻击面的预算 —— 错误体只用来
// 数长度 / 取 sha256 前缀 / 抽 error.code,因此 4 KiB 足够,绝不沿用 50 MiB
// (那会把「一个超大错误响应」变成一次内存分配与一次日志面暴露)。
maxErrorBodyBytes = 4 << 10
)
// imageFetchTimeout 限制一次图像 URL 取回的时长。适配器与消息解析器会
+6 -1
View File
@@ -20,7 +20,12 @@ import (
// 约束单次调用。
const (
cloudMaxResponseBytes = 8 << 20
cloudHTTPTimeout = 180 * time.Second
// maxErrorBodyBytes 是非 2xx 响应(错误体)的读取上限。它与
// cloudMaxResponseBytes 是不同口径:那个是数据面(音频/JSON 响应)的预算,
// 这一条是攻击面的预算 —— 错误体只用来数长度 / 取 sha256 前缀 / 抽
// code 与 request_id,因此 4 KiB 足够,绝不整包读进来。
maxErrorBodyBytes = 4 << 10
cloudHTTPTimeout = 180 * time.Second
)
// HTTPSTT 通过向云端 endpoint POST 一个 JSON 请求体并
+11 -149
View File
@@ -325,7 +325,10 @@ func (d *DashscopeCloner) CloneVoice(ctx context.Context, referenceURL, prefix s
return "", fmt.Errorf("voice: %w: decode: %v", ErrProviderFailed, err)
}
if resp.Message != "" {
return "", fmt.Errorf("voice: %w: %s", ErrProviderFailed, resp.Message)
// 200 + 非空 message 是 DashScope 的「业务级失败」形状。message 原文是
// 上游自由文本(可能回显参考音频的预签名 URL),因此只留形状事实与摘要
// (见 dashscopeOKBodyError),不拼原文。
return "", dashscopeOKBodyError(d.endpoint, body)
}
if resp.Output.VoiceID == "" {
return "", fmt.Errorf("voice: %w: no voice_id in response", ErrProviderFailed)
@@ -371,8 +374,9 @@ func firstNonEmpty(a, b string) string {
}
// dashscopePost POST 一个 JSON 载荷,带上 DashScope bearer 鉴权与可选的
// X-DashScope-SSE 头,返回解码后的响应体。非 200
// 状态会被包装上 DashScope 的错误体。
// X-DashScope-SSE 头,返回解码后的响应体。非 200 状态的错误体**不整包读**:
// 错误路径只读 maxErrorBodyBytes,且正文一个字节都不进 error(见
// classifyDashscopeStatus)。
func dashscopePost(ctx context.Context, client *http.Client, apiKey, url string, payload []byte, sse string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
if err != nil {
@@ -390,52 +394,17 @@ func dashscopePost(ctx context.Context, client *http.Client, apiKey, url string,
return nil, fmt.Errorf("voice: %w: request: %v", ErrProviderFailed, err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
errBody, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes))
return nil, classifyDashscopeStatus(url, resp.StatusCode, resp.Header.Get("Content-Type"), errBody)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, cloudMaxResponseBytes))
if err != nil {
return nil, fmt.Errorf("voice: %w: read: %v", ErrProviderFailed, err)
}
if resp.StatusCode != http.StatusOK {
return nil, classifyDashscopeStatus(resp.StatusCode, body)
}
return body, nil
}
// classifyDashscopeStatus 把一个非 200 的 DashScope 响应翻成**可分类**的失败:鉴权失败 /
// 欠费额度 / 限流各自成一条哨兵链(都以 ErrProviderFailed 为底),其余原样归入
// ErrProviderFailed(切片 IM-ASR 的熔断据此判定「再试是不是白试」)。
//
// 判据只用**状态码 + 错误码/消息里的两类关键字**(阿里云官方口径:`Arrearage`、
// `Throttling.*`、额度上限),绝不匹配自由文本描述——把「格式不支持」误判成「欠费」
// 会让一条被拒的语音关掉整个转写面。
func classifyDashscopeStatus(status int, body []byte) error {
detail := strings.TrimSpace(string(body))
if err := dashscopeClassifiedError(status, detail); err != nil {
return err
}
return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrProviderFailed, status, detail)
}
// dashscopeClassifiedError 给出状态码对应的**分类**失败,无法分类时返回 nil。
func dashscopeClassifiedError(status int, detail string) error {
switch {
case status == http.StatusUnauthorized || status == http.StatusForbidden:
return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrAuthFailed, status, detail)
case status == http.StatusTooManyRequests:
return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrRateLimited, status, detail)
}
// 欠费/额度耗尽:阿里云用 HTTP 400(少数情况 402)配上 Arrearage /
// AllocatedQuotaExceeded 这类错误码,因此按错误码关键字归类(不按状态码)。
lowered := strings.ToLower(detail)
if strings.Contains(lowered, "arrearage") || strings.Contains(lowered, "quota") ||
strings.Contains(lowered, "throttling") {
if strings.Contains(lowered, "throttling") {
return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrRateLimited, status, detail)
}
return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrQuotaExceeded, status, detail)
}
return nil
}
// extractSTTText 从 DashScope ASR 响应中取出转写。ASR
// 的输出形状随模型而异,因此按多个候选路径依次尝试。
func extractSTTText(body []byte) (string, error) {
@@ -488,110 +457,3 @@ func mimeOf(format string) string {
return "audio/wav"
}
}
// dashscopeSTTResponse 是 DashScope ASR 响应 envelope。
type dashscopeSTTResponse struct {
Output struct {
Output struct {
Sentence struct {
Text string `json:"text"`
} `json:"sentence"`
Text string `json:"text"`
} `json:"output"`
Text string `json:"text"`
Choices []struct {
Message struct {
Content []struct {
Text string `json:"text"`
} `json:"content"`
} `json:"message"`
} `json:"choices"`
} `json:"output"`
Text string `json:"text"`
}
// dashscopeTTSResponse 是 DashScope TTS 响应 envelope。
type dashscopeTTSResponse struct {
Output struct {
Audio struct {
URL string `json:"url"`
Data string `json:"data"`
} `json:"audio"`
} `json:"output"`
}
// dashscopeSTTRequest 是 ASR 请求体。
type dashscopeSTTRequest struct {
Model string `json:"model"`
Input dashscopeSTTInput `json:"input"`
Parameters dashscopeSTTParameters `json:"parameters"`
}
// dashscopeSTTInput 承载多模态消息列表。
type dashscopeSTTInput struct {
Messages []dashscopeSTTMessage `json:"messages"`
}
// dashscopeSTTMessage 是一个 ASR 用户回合。
type dashscopeSTTMessage struct {
Role string `json:"role"`
Content []dashscopeSTTContent `json:"content"`
}
// dashscopeSTTContent 是一个音频内容块。
type dashscopeSTTContent struct {
Type string `json:"type"`
InputAudio dashscopeInputAudio `json:"input_audio"`
}
// dashscopeInputAudio 承载 base64 data: URI 音频。
type dashscopeInputAudio struct {
Data string `json:"data"`
}
// dashscopeSTTParameters 调整 ASR 请求。
type dashscopeSTTParameters struct {
Format string `json:"format"`
SampleRate int `json:"sample_rate"`
LanguageHints []string `json:"language_hints,omitempty"`
}
// dashscopeTTSRequest 是 TTS 请求体。
type dashscopeTTSRequest struct {
Model string `json:"model"`
Input dashscopeTTSInput `json:"input"`
}
// dashscopeTTSInput 承载文字 + 音色 + 输出格式。
type dashscopeTTSInput struct {
Text string `json:"text"`
Voice string `json:"voice"`
Format string `json:"format"`
SampleRate int `json:"sample_rate"`
// Instruction 是 CosyVoice 的情绪/语气 instruction;omitempty 使
// 系统音色(必须使用固定 instruction)不携带自由形式的 instruction。
Instruction string `json:"instruction,omitempty"`
}
// dashscopeCloneRequest 是音色注册 create_voice 请求体。
type dashscopeCloneRequest struct {
Model string `json:"model"`
Input dashscopeCloneInput `json:"input"`
}
// dashscopeCloneInput 承载 create_voice 动作参数。
type dashscopeCloneInput struct {
Action string `json:"action"`
TargetModel string `json:"target_model"`
Prefix string `json:"prefix"`
URL string `json:"url"`
}
// dashscopeCloneResponse 是音色注册响应 envelope。顶层
// message 表示模型级错误——即使 HTTP 200 也会暴露。
type dashscopeCloneResponse struct {
Output struct {
VoiceID string `json:"voice_id"`
} `json:"output"`
Message string `json:"message"`
}
@@ -0,0 +1,136 @@
package voice
// 本文件是 DashScope 的「非 2xx 响应 → error」收口,与 M41 的 LLM / embedding /
// vision 三处同口径:走 internal/runtime/trace.UpstreamError(M40 共享 helper),
// 不自己再写一份「截一段响应体拼进 error」。
//
// 这一家的敏感面是 **TTS 的 input = 她要念的那段话**(与 emotionTTS 同型):
// DashScope 的 4xx 错误体里既有 `code` / `request_id`,也有自由文本 `message`,
// 而 message 会回显请求内容。因此正文只留在内存里做分类判据,出镜的只有
// 状态、体积摘要以及两个受控字段(code / request_id)。
import (
"encoding/json"
"fmt"
"net/http"
"strings"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// maxErrorTokenBytes 是 error.code / request_id 允许出镜的长度上限(见
// dashscopeErrorToken)。
const maxErrorTokenBytes = 64
// classifyDashscopeStatus 把一个非 200 的 DashScope 响应翻成**可分类**的失败:鉴权失败 /
// 欠费额度 / 限流各自成一条哨兵链(都以 ErrProviderFailed 为底),其余原样归入
// ErrProviderFailed(切片 IM-ASR 的熔断据此判定「再试是不是白试」)。
//
// 判据只用**状态码 + 错误码/消息里的两类关键字**(阿里云官方口径:`Arrearage`、
// `Throttling.*`、额度上限),绝不匹配自由文本描述——把「格式不支持」误判成「欠费」
// 会让一条被拒的语音关掉整个转写面。
//
// 🔴 错误文案里**没有正文**:响应体只用来(a)在内存里做上面的关键字分类、
// (b)数长度 / 取 sha256 前缀、(c)解出 code 与 request_id。出镜的只有
// status / contentType / bodyLen / sha256 / code / request_id —— TTS 的请求
// input 就是**她要念的那段话**,把它(或回显它的 message)拼进 error 等于
// 把台词写进日志。
//
// 双 %w:errors.Is 既能命中分类哨兵(熔断判据),也能命中 trace.ErrUpstream。
func classifyDashscopeStatus(endpoint string, status int, contentType string, body []byte) error {
sentinel := dashscopeSentinelFor(status, strings.ToLower(string(body)))
fields := decodeDashscopeErrorFields(body)
info, ok := trace.UpstreamError(trace.HostOf(endpoint), status, contentType, body).(*trace.UpstreamErrorInfo)
if !ok {
// helper 的构造是固定的,走不到这里;真走到也只少一条诊断。
return fmt.Errorf("voice: %w", sentinel)
}
info.Code = dashscopeErrorToken(fields.code)
if rid := dashscopeErrorToken(fields.requestID); rid != "" {
return fmt.Errorf("voice: %w: request_id=%s: %w", sentinel, rid, info)
}
return fmt.Errorf("voice: %w: %w", sentinel, info)
}
// dashscopeSentinelFor 给出状态码/错误码对应的**分类**哨兵,无法分类时返回
// ErrProviderFailed。loweredBody 只用于关键字判定,绝不外泄 —— 分类是「再试
// 是不是白试」的判据,不是一条给人看的描述。
func dashscopeSentinelFor(status int, loweredBody string) error {
switch status {
case http.StatusUnauthorized, http.StatusForbidden:
return ErrAuthFailed
case http.StatusTooManyRequests:
return ErrRateLimited
}
// 欠费/额度耗尽:阿里云用 HTTP 400(少数情况 402)配上 Arrearage /
// AllocatedQuotaExceeded 这类错误码,因此按错误码关键字归类(不按状态码)。
switch {
case strings.Contains(loweredBody, "throttling"):
return ErrRateLimited
case strings.Contains(loweredBody, "arrearage"), strings.Contains(loweredBody, "quota"):
return ErrQuotaExceeded
}
return ErrProviderFailed
}
// dashscopeErrorFields 是 DashScope 非 2xx 响应体里**唯一**允许出镜的两个字段。
// message 不在此列:它是上游自由文本,TTS 场景下可能回显她要念的文本。
type dashscopeErrorFields struct {
code string
requestID string
}
// decodeDashscopeErrorFields 尽力解出 code / request_id。不是 JSON(纯文本或
// 被截断的报文)⇒ 两个字段都是空串,错误里只剩 status/体积/摘要 —— 正文一个
// 字节都不留。
func decodeDashscopeErrorFields(body []byte) dashscopeErrorFields {
var wire struct {
Code string `json:"code"`
RequestID string `json:"request_id"`
}
if err := json.Unmarshal(body, &wire); err != nil {
return dashscopeErrorFields{}
}
return dashscopeErrorFields{code: wire.Code, requestID: wire.RequestID}
}
// dashscopeOKBodyError 处理 DashScope 的「HTTP 200 但业务失败」形状:音色复刻
// (create_voice)用顶层 message 报错,因此**不走** classifyDashscopeStatus ——
// 200 这条路必须单独收口,否则它就成了这一族里唯一还漏的缺口。
//
// 只出镜一个**形状事实**(打到了、但 200 响应是个业务失败)加上体积摘要;
// message 原文一律不进 error:它是上游自由文本,而这条请求的 input.url 是参考
// 音频的公开地址(可能是预签名 URL,带凭据),上游报错经常把它整段回显出来。
// 病因分类保留在 ErrProviderFailed(再试也不会好,与既有的熔断口径一致);
// 确实带回来的 code/type 也只以令牌形态出镜(过 dashscopeErrorToken)。
func dashscopeOKBodyError(endpoint string, body []byte) error {
info, ok := trace.UpstreamError(trace.HostOf(endpoint), http.StatusOK, "", body).(*trace.UpstreamErrorInfo)
if !ok {
// helper 的构造是固定的,走不到这里;真走到也只少一条诊断。
return fmt.Errorf("voice: %w: dashscope reported a business failure on a 200 response", ErrProviderFailed)
}
info.Code = dashscopeErrorToken(info.Code)
info.Type = dashscopeErrorToken(info.Type)
return fmt.Errorf("voice: %w: dashscope reported a business failure on a 200 response: %w",
ErrProviderFailed, info)
}
// dashscopeErrorToken 只放行「像协议令牌」的短串(字母、数字与 `_ . -`):code
// 在 DashScope 是枚举(InvalidParameter / Throttling.RateQuota / Arrearage …),
// request_id 是十六进制/UUID 形状。两者都是自由文本面,不像令牌的一律丢弃 ——
// 宁可少一条分类提示,不让散文借道出镜(同口径见 runtime/model 的 safeErrorToken)。
func dashscopeErrorToken(s string) string {
s = strings.TrimSpace(s)
if s == "" || len(s) > maxErrorTokenBytes {
return ""
}
for _, r := range s {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
case r == '_' || r == '.' || r == '-':
default:
return ""
}
}
return s
}
@@ -115,6 +115,40 @@ func TestDashscopeSTT_Non200(t *testing.T) {
}
}
// TestDashscopeTTS_Non200DropsSpokenText 覆盖 M41 的 DashScope 一处:TTS 请求的
// input 就是**她要念的那段话**,而 DashScope 的 4xx 错误体会把它(或回显它的
// message)带回来。旧实现把整包响应体(cloudMaxResponseBytes = 8 MiB,无截断)
// 拼进 error,error 又被语音链打日志。
//
// 断言「文本里搜不到台词」,同时保留两类可诊断字段:分类哨兵链与 code /
// request_id(DashScope 报障时确实要看这两个)。
func TestDashscopeTTS_Non200DropsSpokenText(t *testing.T) {
t.Parallel()
const spoken = "SENTINEL-SPOKEN-TEXT-1e4f"
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"code":"InvalidParameter","message":"text: '` + spoken +
`' is invalid","request_id":"req-0123456789abcdef"}`))
}))
defer srv.Close()
tts := NewDashscopeTTS(config.VoiceConfig{DashscopeBaseURL: srv.URL, DashscopeAPIKey: "sk"})
_, err := tts.TTS(context.Background(), spoken)
if !errors.Is(err, ErrProviderFailed) {
t.Fatalf("err = %v, want ErrProviderFailed", err)
}
if strings.Contains(err.Error(), spoken) {
t.Errorf("error text echoes the text to be spoken: %v", err)
}
if !strings.Contains(err.Error(), "InvalidParameter") {
t.Errorf("err = %v, want the error code kept for triage", err)
}
if !strings.Contains(err.Error(), "req-0123456789abcdef") {
t.Errorf("err = %v, want the request id kept for provider support", err)
}
}
func TestDashscopeTTS_URL(t *testing.T) {
t.Parallel()
audioBytes := []byte("RIFFxxxxWAVE")
@@ -0,0 +1,111 @@
package voice
// DashScope 的线格式类型(请求体 / 响应 envelope)。它们只描述 JSON 形状,
// 不含任何逻辑,因此与适配器行为分开成文件。
// dashscopeSTTResponse 是 DashScope ASR 响应 envelope。
type dashscopeSTTResponse struct {
Output struct {
Output struct {
Sentence struct {
Text string `json:"text"`
} `json:"sentence"`
Text string `json:"text"`
} `json:"output"`
Text string `json:"text"`
Choices []struct {
Message struct {
Content []struct {
Text string `json:"text"`
} `json:"content"`
} `json:"message"`
} `json:"choices"`
} `json:"output"`
Text string `json:"text"`
}
// dashscopeTTSResponse 是 DashScope TTS 响应 envelope。
type dashscopeTTSResponse struct {
Output struct {
Audio struct {
URL string `json:"url"`
Data string `json:"data"`
} `json:"audio"`
} `json:"output"`
}
// dashscopeSTTRequest 是 ASR 请求体。
type dashscopeSTTRequest struct {
Model string `json:"model"`
Input dashscopeSTTInput `json:"input"`
Parameters dashscopeSTTParameters `json:"parameters"`
}
// dashscopeSTTInput 承载多模态消息列表。
type dashscopeSTTInput struct {
Messages []dashscopeSTTMessage `json:"messages"`
}
// dashscopeSTTMessage 是一个 ASR 用户回合。
type dashscopeSTTMessage struct {
Role string `json:"role"`
Content []dashscopeSTTContent `json:"content"`
}
// dashscopeSTTContent 是一个音频内容块。
type dashscopeSTTContent struct {
Type string `json:"type"`
InputAudio dashscopeInputAudio `json:"input_audio"`
}
// dashscopeInputAudio 承载 base64 data: URI 音频。
type dashscopeInputAudio struct {
Data string `json:"data"`
}
// dashscopeSTTParameters 调整 ASR 请求。
type dashscopeSTTParameters struct {
Format string `json:"format"`
SampleRate int `json:"sample_rate"`
LanguageHints []string `json:"language_hints,omitempty"`
}
// dashscopeTTSRequest 是 TTS 请求体。
type dashscopeTTSRequest struct {
Model string `json:"model"`
Input dashscopeTTSInput `json:"input"`
}
// dashscopeTTSInput 承载文字 + 音色 + 输出格式。
type dashscopeTTSInput struct {
Text string `json:"text"`
Voice string `json:"voice"`
Format string `json:"format"`
SampleRate int `json:"sample_rate"`
// Instruction 是 CosyVoice 的情绪/语气 instruction;omitempty 使
// 系统音色(必须使用固定 instruction)不携带自由形式的 instruction。
Instruction string `json:"instruction,omitempty"`
}
// dashscopeCloneRequest 是音色注册 create_voice 请求体。
type dashscopeCloneRequest struct {
Model string `json:"model"`
Input dashscopeCloneInput `json:"input"`
}
// dashscopeCloneInput 承载 create_voice 动作参数。
type dashscopeCloneInput struct {
Action string `json:"action"`
TargetModel string `json:"target_model"`
Prefix string `json:"prefix"`
URL string `json:"url"`
}
// dashscopeCloneResponse 是音色注册响应 envelope。顶层
// message 表示模型级错误——即使 HTTP 200 也会暴露。
type dashscopeCloneResponse struct {
Output struct {
VoiceID string `json:"voice_id"`
} `json:"output"`
Message string `json:"message"`
}
@@ -151,6 +151,35 @@ func TestDashscopeCloner_ModelError(t *testing.T) {
}
}
// TestDashscopeCloner_ModelErrorHidesMessage 覆盖 M41 的最后一条同族缺口:
// HTTP **200** 的响应里用顶层 message 报错(200 不会走非 2xx 的收口),而这句话是
// 上游自由文本 —— create_voice 的 input.url 是参考音频的公开地址(可能是预签名
// URL,带凭据),上游经常把它原样回显在 message 里。
//
// 两个断言缺一不可:① 失败必须**仍被识别**(ErrProviderFailed,熔断口径不变);
// ② error 文本里搜不到哨兵(旧实现 `%s resp.Message` 会整句带出)。
func TestDashscopeCloner_ModelErrorHidesMessage(t *testing.T) {
t.Parallel()
const echoed = "SENTINEL-CLONE-MESSAGE-3c7b"
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"message":"cannot fetch ` + echoed + `: signature expired","request_id":"req-abc"}`))
}))
defer srv.Close()
cloner := NewDashscopeCloner(config.VoiceConfig{DashscopeBaseURL: srv.URL, DashscopeAPIKey: "sk"})
_, err := cloner.CloneVoice(context.Background(), "https://example.com/ref.wav", "x")
if !errors.Is(err, ErrProviderFailed) {
t.Fatalf("err = %v, want ErrProviderFailed", err)
}
if strings.Contains(err.Error(), echoed) {
t.Errorf("error text carries the provider message: %v", err)
}
if !strings.Contains(err.Error(), "200") {
t.Errorf("err = %v, want the 200-on-business-failure fact kept", err)
}
}
// TestDashscopeTTS_Instruction_ClonedVoice 断言复刻音色会在请求 input 中
// 携带配置的 instruction。
func TestDashscopeTTS_Instruction_ClonedVoice(t *testing.T) {