diff --git a/internal/runtime/embedding/api.go b/internal/runtime/embedding/api.go index b3f1559..2340316 100644 --- a/internal/runtime/embedding/api.go +++ b/internal/runtime/embedding/api.go @@ -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 diff --git a/internal/runtime/embedding/api_test.go b/internal/runtime/embedding/api_test.go index f288a13..aaaba03 100644 --- a/internal/runtime/embedding/api_test.go +++ b/internal/runtime/embedding/api_test.go @@ -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() diff --git a/internal/runtime/model/anthropic.go b/internal/runtime/model/anthropic.go index 40d9822..ca92fa5 100644 --- a/internal/runtime/model/anthropic.go +++ b/internal/runtime/model/anthropic.go @@ -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 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率, diff --git a/internal/runtime/model/anthropic_stream.go b/internal/runtime/model/anthropic_stream.go index ee5636b..820199c 100644 --- a/internal/runtime/model/anthropic_stream.go +++ b/internal/runtime/model/anthropic_stream.go @@ -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 diff --git a/internal/runtime/model/anthropic_test.go b/internal/runtime/model/anthropic_test.go index 3881786..976be05 100644 --- a/internal/runtime/model/anthropic_test.go +++ b/internal/runtime/model/anthropic_test.go @@ -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"}) diff --git a/internal/runtime/model/openai.go b/internal/runtime/model/openai.go index a456547..0612259 100644 --- a/internal/runtime/model/openai.go +++ b/internal/runtime/model/openai.go @@ -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 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率, diff --git a/internal/runtime/model/openai_test.go b/internal/runtime/model/openai_test.go index 40dd960..67695cf 100644 --- a/internal/runtime/model/openai_test.go +++ b/internal/runtime/model/openai_test.go @@ -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{}) diff --git a/internal/runtime/model/upstream.go b/internal/runtime/model/upstream.go new file mode 100644 index 0000000..f272561 --- /dev/null +++ b/internal/runtime/model/upstream.go @@ -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 +} diff --git a/internal/runtime/multimodal/vision/adapter.go b/internal/runtime/multimodal/vision/adapter.go index 74ba704..e9d3563 100644 --- a/internal/runtime/multimodal/vision/adapter.go +++ b/internal/runtime/multimodal/vision/adapter.go @@ -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) } diff --git a/internal/runtime/multimodal/vision/adapter_test.go b/internal/runtime/multimodal/vision/adapter_test.go index 4b4dcfb..79bd481 100644 --- a/internal/runtime/multimodal/vision/adapter_test.go +++ b/internal/runtime/multimodal/vision/adapter_test.go @@ -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) + } +} diff --git a/internal/runtime/multimodal/vision/image.go b/internal/runtime/multimodal/vision/image.go index c0e13de..68164bb 100644 --- a/internal/runtime/multimodal/vision/image.go +++ b/internal/runtime/multimodal/vision/image.go @@ -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 取回的时长。适配器与消息解析器会 diff --git a/internal/runtime/multimodal/voice/cloud.go b/internal/runtime/multimodal/voice/cloud.go index 032054a..a8fdcdf 100644 --- a/internal/runtime/multimodal/voice/cloud.go +++ b/internal/runtime/multimodal/voice/cloud.go @@ -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 请求体并 diff --git a/internal/runtime/multimodal/voice/dashscope.go b/internal/runtime/multimodal/voice/dashscope.go index 511eeb6..de20e40 100644 --- a/internal/runtime/multimodal/voice/dashscope.go +++ b/internal/runtime/multimodal/voice/dashscope.go @@ -371,8 +371,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 +391,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 +454,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"` -} diff --git a/internal/runtime/multimodal/voice/dashscope_error.go b/internal/runtime/multimodal/voice/dashscope_error.go new file mode 100644 index 0000000..858d345 --- /dev/null +++ b/internal/runtime/multimodal/voice/dashscope_error.go @@ -0,0 +1,115 @@ +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} +} + +// 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 +} diff --git a/internal/runtime/multimodal/voice/dashscope_test.go b/internal/runtime/multimodal/voice/dashscope_test.go index 592e9f8..a4d3e9e 100644 --- a/internal/runtime/multimodal/voice/dashscope_test.go +++ b/internal/runtime/multimodal/voice/dashscope_test.go @@ -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") diff --git a/internal/runtime/multimodal/voice/dashscope_wire.go b/internal/runtime/multimodal/voice/dashscope_wire.go new file mode 100644 index 0000000..e0ee907 --- /dev/null +++ b/internal/runtime/multimodal/voice/dashscope_wire.go @@ -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"` +}