fix(runtime): 上游错误体统一走 trace.UpstreamError,正文不进 error
M41 的「LLM / embedding / 视觉 上游错误体」一族(M38 #5 #6 #7 #8 #9 #14): 非 2xx 时把响应体截断后拼进 error,而 error 会被 chat run / loop step / retrieve_fallback 打日志,等于把她的提问、记忆正文、要念的台词写进日志。 - model:openai/anthropic 两处 upstreamError 收口到 upstream.go 的 providerUpstreamError;anthropic 流内 error 事件的 evt.Error.Message (原本无上限)整段丢弃,只保留令牌形状的 type - embedding:readErrorBody 删除,改走 helper(input 就是她的记忆正文) - vision:错误路径不再沿用取图的 50 MiB 上限,改为 4 KiB 有界读且不拼原文 - voice:classifyDashscopeStatus 只解 code/request_id,正文只留在内存里做 分类判据;错误体读取同样改为 4 KiB 有界 - 测试:五个上游各一条「日志/error 里搜不到哨兵正文」用例,且能变异打红
This commit is contained in:
1 parent
46a6da13e0
commit
66dda084cd
16 files changed
+570
-183
No files matched your search
@@ -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 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"})
|
||||
|
||||
@@ -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 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
|
||||
|
||||
@@ -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{})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in new issue
Block a user