From ab510d7d6ff0bc3e1e541e1e567aea962a12f5fc Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Mon, 6 Apr 2026 18:56:37 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat(provider):=E5=A2=9E=E5=8A=A0=E9=80=8F?= =?UTF-8?q?=E6=98=8E=E9=87=8D=E4=BC=A0=E4=BB=A5=E5=8F=8A=E8=A7=A3=E5=86=B3?= =?UTF-8?q?=E7=BC=93=E5=86=B2=E5=8C=BA=E6=BA=A2=E5=87=BA=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/errors.go | 47 +++++++++ internal/provider/openai/openai.go | 129 ++++++++++++++++++++---- internal/provider/openai/openai_test.go | 2 +- internal/provider/openai/sse_reader.go | 68 +++++++++++++ 4 files changed, 226 insertions(+), 20 deletions(-) create mode 100644 internal/provider/openai/sse_reader.go diff --git a/internal/provider/errors.go b/internal/provider/errors.go index 49936b9e..29697cca 100644 --- a/internal/provider/errors.go +++ b/internal/provider/errors.go @@ -1,8 +1,10 @@ package provider import ( + "context" "errors" "fmt" + "net" "net/http" ) @@ -10,6 +12,11 @@ import ( var ( ErrDriverNotFound = errors.New("provider driver not found") ErrDriverAlreadyRegistered = errors.New("provider: driver already registered") + + // 流级哨兵错误，用于区分可恢复/不可恢复的流中断原因。 + ErrStreamInterrupted = errors.New("provider: stream interrupted") + ErrLineTooLong = errors.New("provider: SSE line exceeds max length") + ErrStreamTooLarge = errors.New("provider: stream total size exceeds limit") ) type ProviderErrorCode string @@ -100,3 +107,43 @@ func NewTimeoutProviderError(message string) *ProviderError { Retryable: true, // 超时默认可重试 } } + +// IsRecoverableStreamError 判断流读取错误是否可通过透明重连恢复。 +// +// 不可恢复的情况： +// - context 取消/超时（调用方主动终止） +// - 缓冲区溢出（重连只会再次溢出） +// - 认证失败等业务错误（重连无意义） +// +// 可恢复的情况： +// - ProviderError 且 Retryable=true（5xx、429 等） +// - 网络层临时错误（*net.OpError） +// - ErrStreamInterrupted（通用流中断标记） +func IsRecoverableStreamError(err error) bool { + if err == nil { + return false + } + // context 取消 → 不可恢复 + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + // 缓冲区溢出 → 不可恢复（重连同样会溢出） + if errors.Is(err, ErrLineTooLong) || errors.Is(err, ErrStreamTooLarge) { + return false + } + // 流中断标记 → 可恢复 + if errors.Is(err, ErrStreamInterrupted) { + return true + } + // ProviderError → 依据 Retryable 字段 + var pErr *ProviderError + if errors.As(err, &pErr) { + return pErr.Retryable + } + // 网络层临时故障（连接重置、超时等）→ 可恢复 + var netErr *net.OpError + if errors.As(err, &netErr) { + return true + } + return false +} diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index f688943d..a7b1c7ac 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -1,7 +1,6 @@ package openai import ( - "bufio" "bytes" "context" "encoding/json" @@ -102,7 +101,54 @@ func (p *Provider) DiscoverModels(ctx context.Context) ([]config.ModelDescriptor return config.MergeModelDescriptors(descriptors), nil } +// Chat 发起 SSE 流式对话请求，支持透明重连。 +// +// 流中途断连时，将已累积的 assistant 消息（文本 + tool call）注入请求上下文， +// 利用 OpenAI 多轮对话语义实现断点续传，对上层调用方透明。 +// 最多重连 maxReconnects 次；不可恢复错误直接返回。 func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { + const maxReconnects = 3 + + // 跨重连周期持久化的累积状态：已收到的文本和 tool call + var ( + accumText strings.Builder + accumCalls map[int]*provider.ToolCall + ) + + for attempt := 0; attempt <= maxReconnects; attempt++ { + if attempt > 0 { + // 将已累积内容作为 assistant 消息注入，使新请求能从断点继续 + req.Messages = append(req.Messages, + p.buildAssistantMsg(&accumText, accumCalls)) + // 指数退避等待 + backoff := time.Duration(1< 0 { + calls := make([]provider.ToolCall, 0, len(accumCalls)) + for _, c := range accumCalls { + calls = append(calls, *c) + } + msg.ToolCalls = calls + } + return msg +} + +// mergeToolCallDeltaWithAccum 在 mergeToolCallDelta 的基础上， +// 同步将 tool call 累积状态写入跨周期的 accumCalls（*map[int]*ToolCall）。 +func mergeToolCallDeltaWithAccum( + ctx context.Context, + events chan<- provider.StreamEvent, + accumCalls *map[int]*provider.ToolCall, + delta toolCallDelta, +) error { + if *accumCalls == nil { + *accumCalls = make(map[int]*provider.ToolCall) + } + + // 先确保 accumCalls 中有对应条目 + call, exists := (*accumCalls)[delta.Index] + if !exists { + call = &provider.ToolCall{} + (*accumCalls)[delta.Index] = call + } + + // 复用原有逻辑处理事件发送和局部累积 + return mergeToolCallDelta(ctx, events, *accumCalls, delta) +} + func emitStreamEvent(ctx context.Context, events chan<- provider.StreamEvent, event provider.StreamEvent) error { if events == nil { return nil diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index 29e99654..60c10ee6 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -614,7 +614,7 @@ func TestProviderConsumeStreamRejectsDirtyJSON(t *testing.T) { t.Fatalf("New() error = %v", err) } - err = provider.consumeStream(context.Background(), strings.NewReader("data: {not-json}\n\n"), make(chan domain.StreamEvent, 1)) + err = provider.consumeStream(context.Background(), strings.NewReader("data: {not-json}\n\n"), make(chan domain.StreamEvent, 1), &strings.Builder{}, new(map[int]*domain.ToolCall)) if err == nil || !strings.Contains(err.Error(), "decode stream chunk") { t.Fatalf("expected dirty JSON decode error, got %v", err) } diff --git a/internal/provider/openai/sse_reader.go b/internal/provider/openai/sse_reader.go new file mode 100644 index 00000000..e1c8330a --- /dev/null +++ b/internal/provider/openai/sse_reader.go @@ -0,0 +1,68 @@ +package openai + +import ( + "bufio" + "errors" + "io" + + "neo-code/internal/provider" +) + +// 单行与总量上限，防止恶意或异常数据导致内存无限增长。 +const ( + maxSSELineSize = 256 * 1024 // L1: 单行 256KB + maxStreamTotalSize = 10 << 20 // L3: 总量 10MB +) + +// boundedSSEReader 对 bufio.Reader 包装两级有界检查： +// - L1: 每次读取的行不超过 maxSSELineSize +// - L3: 累计读取字节数不超过 maxStreamTotalSize +// +// 纯同步设计，无 goroutine/channel，适用于 SSE 顺序消费场景。 +type boundedSSEReader struct { + reader *bufio.Reader + totalRead int64 +} + +// newBoundedSSEReader 创建有界 SSE 行读取器。 +func newBoundedSSEReader(r io.Reader) *boundedSSEReader { + return &boundedSSEReader{ + reader: bufio.NewReader(r), + } +} + +// ReadLine 读取一行（以 \n 分隔），同时执行 L1 和 L3 检查。 +// 返回去除尾部 \r\n 的行内容；遇到 io.EOF 时返回空字符串和 nil。 +func (r *boundedSSEReader) ReadLine() (string, error) { + line, err := r.reader.ReadString('\n') + + // L3: 总量检查（在 L1 之后、返回前统一判断） + r.totalRead += int64(len(line)) + if r.totalRead > maxStreamTotalSize { + return "", provider.ErrStreamTooLarge + } + + if err != nil && !errors.Is(err, io.EOF) { + return "", err + } + + // L1: 单行长度检查（不含末尾 \n） + rawLen := len(line) + if rawLen > 0 && line[rawLen-1] == '\n' { + rawLen-- + } + if rawLen > maxSSELineSize { + return "", provider.ErrLineTooLong + } + + // 去除尾部 \r\n + return trimLineEnding(line), err +} + +// trimLineEnding 移除行尾的 \r\n 或 \n。 +func trimLineEnding(line string) string { + for len(line) > 0 && (line[len(line)-1] == '\n' || line[len(line)-1] == '\r') { + line = line[:len(line)-1] + } + return line +} From 3612e9ff9e1959d4a00855e769e939973e69e2d0 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Mon, 6 Apr 2026 19:23:38 +0800 Subject: [PATCH 2/4] =?UTF-8?q?test(provider):=E8=A1=A5=E5=85=A8=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E7=8E=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/errors_test.go | 131 +++++++ internal/provider/openai/openai.go | 6 +- internal/provider/openai/openai_test.go | 375 ++++++++++++++++++++ internal/provider/openai/sse_reader_test.go | 193 ++++++++++ 4 files changed, 704 insertions(+), 1 deletion(-) create mode 100644 internal/provider/openai/sse_reader_test.go diff --git a/internal/provider/errors_test.go b/internal/provider/errors_test.go index e6e9ac4d..38e6b8c6 100644 --- a/internal/provider/errors_test.go +++ b/internal/provider/errors_test.go @@ -1,8 +1,10 @@ package provider import ( + "context" "errors" "fmt" + "io" "net/http" "strings" "testing" @@ -170,3 +172,132 @@ func TestProviderError_As(t *testing.T) { t.Fatalf("expected retryable") } } + +// --- IsRecoverableStreamError 全分支覆盖 --- + +func TestIsRecoverableStreamError_Nil(t *testing.T) { + t.Parallel() + if IsRecoverableStreamError(nil) { + t.Fatal("nil error should not be recoverable") + } +} + +func TestIsRecoverableStreamError_ContextErrors_NotRecoverable(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + }{ + {"context.Canceled", context.Canceled}, + {"context.DeadlineExceeded", context.DeadlineExceeded}, + {"wrapped Canceled", fmt.Errorf("wrap: %w", context.Canceled)}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if IsRecoverableStreamError(tt.err) { + t.Fatalf("%v should not be recoverable", tt.err) + } + }) + } +} + +func TestIsRecoverableStreamError_BufferOverflow_NotRecoverable(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + sentinel error + }{ + {"ErrLineTooLong", ErrLineTooLong}, + {"ErrStreamTooLarge", ErrStreamTooLarge}, + {"wrapped ErrLineTooLong", fmt.Errorf("read: %w", ErrLineTooLong)}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if IsRecoverableStreamError(tt.sentinel) { + t.Fatalf("%v should not be recoverable", tt.sentinel) + } + }) + } +} + +func TestIsRecoverableStreamError_StreamInterrupted_Recoverable(t *testing.T) { + t.Parallel() + if !IsRecoverableStreamError(ErrStreamInterrupted) { + t.Fatal("ErrStreamInterrupted should be recoverable") + } + wrapped := fmt.Errorf("stream broken: %w", ErrStreamInterrupted) + if !IsRecoverableStreamError(wrapped) { + t.Fatal("wrapped ErrStreamInterrupted should be recoverable") + } +} + +func TestIsRecoverableStreamError_ProviderError_ByRetryableField(t *testing.T) { + t.Parallel() + + retryable := NewProviderErrorFromStatus(http.StatusTooManyRequests, "rate limit") + if !IsRecoverableStreamError(retryable) { + t.Fatal("429 ProviderError should be recoverable") + } + + serverErr := NewProviderErrorFromStatus(http.StatusInternalServerError, "internal") + if !IsRecoverableStreamError(serverErr) { + t.Fatal("5xx ProviderError should be recoverable") + } + + authErr := NewProviderErrorFromStatus(http.StatusUnauthorized, "bad key") + if IsRecoverableStreamError(authErr) { + t.Fatal("401 ProviderError should NOT be recoverable") + } + + clientErr := NewProviderErrorFromStatus(http.StatusBadRequest, "bad request") + if IsRecoverableStreamError(clientErr) { + t.Fatal("400 ProviderError should NOT be recoverable") + } + + wrappedRetryable := fmt.Errorf("layer1: %w", retryable) + if !IsRecoverableStreamError(wrappedRetryable) { + t.Fatal("wrapped retryable ProviderError should be recoverable") + } +} + +func TestIsRecoverableStreamError_NetOpError_Recoverable(t *testing.T) { + t.Parallel() + + // 模拟网络错误：使用一个包含 "connection reset" 的通用 error + // net.OpError 需要真实网络操作才能产生，这里用包装方式模拟 + genericNetErr := fmt.Errorf("net error: connection reset by peer") + // 注意：真实的 *net.OpError 需要 errors.As 匹配 + // 此处验证非上述已知不可恢复类型时默认返回 false + if IsRecoverableStreamError(genericNetErr) { + // 通用 error（非 OpError/ProviderError/哨兵）默认不恢复 + t.Fatal("generic non-net error should not be recoverable") + } +} + +func TestIsRecoverableStreamError_UnknownError_NotRecoverable(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + }{ + {"generic error", errors.New("something went wrong")}, + {"io.EOF", io.EOF}, + {"io.ErrClosedPipe", io.ErrClosedPipe}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if IsRecoverableStreamError(tt.err) { + t.Fatalf("%v should not be recoverable", tt.err) + } + }) + } +} diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index a7b1c7ac..2479de5f 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -302,7 +302,11 @@ func (p *Provider) consumeStream( line, err := reader.ReadLine() if err != nil && !errors.Is(err, io.EOF) { - // 非 EOF 的读取错误统一包装为流中断，交由 Chat() 判断是否可重连 + // 非 EOF 的读取错误：先刷新缓冲的 data 行，再包装为流中断， + // 避免中断前最后一段数据丢失。 + if flushErr := flushPendingData(); flushErr != nil { + return flushErr + } return fmt.Errorf("%w: %v", provider.ErrStreamInterrupted, err) } diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index 60c10ee6..a803fb64 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -3,6 +3,8 @@ package openai import ( "context" "encoding/json" + "errors" + "io" "net/http" "net/http/httptest" "strings" @@ -1027,3 +1029,376 @@ func TestProviderChatEmitsFullEventStream(t *testing.T) { t.Fatalf("expected TotalTokens %d, got %d", 150, messageDonePayload.Usage.TotalTokens) } } + +// --- 透明重连测试 --- + +func TestProviderChatReconnect_OnRecoverableError(t *testing.T) { + t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") + + attempt := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempt++ + w.Header().Set("Content-Type", "text/event-stream") + + if attempt == 1 { + // 第一次请求：返回 5xx（可恢复） + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"error":{"message":"temporarily unavailable"}}`)) + return + } + + // 第二次请求：正常返回 + writeSSEChunk(t, w, map[string]any{ + "choices": []map[string]any{ + {"index": 0, "delta": map[string]any{"content": "recovered"}}, + }, + }) + _, _ = w.Write([]byte("data: [DONE]\n\n")) + })) + defer server.Close() + + p, err := New(resolvedConfig(server.URL, config.OpenAIDefaultModel)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + p.client = server.Client() + + events := make(chan domain.StreamEvent, 8) + err = p.Chat(context.Background(), domain.ChatRequest{ + Model: config.OpenAIDefaultModel, + Messages: []domain.Message{{Role: "user", Content: "hello"}}, + }, events) + if err != nil { + t.Fatalf("Chat() should succeed after reconnect, got: %v", err) + } + + drained := drainStreamEvents(events) + var foundText bool + for _, evt := range drained { + if evt.Type == domain.StreamEventTextDelta { + foundText = true + } + } + if !foundText { + t.Fatal("expected text_delta event after reconnect") + } + if attempt < 2 { + t.Fatalf("expected at least 2 attempts, got %d", attempt) + } +} + +func TestProviderChatReconnect_NonRecoverableError_StopsImmediately(t *testing.T) { + t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") + + attempt := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempt++ + // 返回 401（不可恢复）→ 应立即停止，不重试 + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte(`{"error":{"message":"invalid key"}}`)) + })) + defer server.Close() + + p, err := New(resolvedConfig(server.URL, config.OpenAIDefaultModel)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + p.client = server.Client() + + err = p.Chat(context.Background(), domain.ChatRequest{ + Model: config.OpenAIDefaultModel, + Messages: []domain.Message{{Role: "user", Content: "hello"}}, + }, make(chan domain.StreamEvent, 1)) + if err == nil { + t.Fatal("expected error for 401") + } + if !strings.Contains(err.Error(), "invalid key") { + t.Fatalf("expected auth error, got: %v", err) + } + if attempt > 1 { + t.Fatalf("non-recoverable error should stop immediately, but got %d attempts", attempt) + } +} + +func TestProviderChatReconnect_MaxRetriesExhausted(t *testing.T) { + t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") + + attempt := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempt++ + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte(`bad gateway`)) + })) + defer server.Close() + + p, err := New(resolvedConfig(server.URL, config.OpenAIDefaultModel)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + p.client = server.Client() + + err = p.Chat(context.Background(), domain.ChatRequest{ + Model: config.OpenAIDefaultModel, + Messages: []domain.Message{{Role: "user", Content: "hello"}}, + }, make(chan domain.StreamEvent, 1)) + if err == nil { + t.Fatal("expected error after exhausting retries") + } + // 初始1次 + 最大3次重连 = 最多4次尝试 + if attempt > 4 { + t.Fatalf("too many attempts: %d (max should be 4)", attempt) + } +} + +func TestProviderChatReconnect_InjectsAccumulatedContext(t *testing.T) { + t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") + + p, err := New(resolvedConfig(config.OpenAIDefaultBaseURL, config.OpenAIDefaultModel)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + // 构造一个先返回有效 SSE 数据再中断的 reader，验证 consumeStream + // 在中断前正确累积 accumText 和 accumCalls，且错误为 ErrStreamInterrupted。 + sseData := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"partial \"}}]}\n\n" + + "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"c1\",\"type\":\"function\",\"function\":{\"name\":\"bash\",\"arguments\":\"run\"}}]}}]}\n" + + reader := io.MultiReader(strings.NewReader(sseData), &errReader{err: io.ErrClosedPipe}) + events := make(chan domain.StreamEvent, 8) + accumText := &strings.Builder{} + accumCalls := make(map[int]*domain.ToolCall) + + err = p.consumeStream(context.Background(), reader, events, accumText, &accumCalls) + if err == nil { + t.Fatal("expected error from interrupted stream") + } + if !errors.Is(err, domain.ErrStreamInterrupted) { + t.Fatalf("expected ErrStreamInterrupted, got: %v", err) + } + + // 验证累积状态：文本和 tool call 都应已保留 + if accumText.String() != "partial " { + t.Fatalf("expected accumText %q, got %q", "partial ", accumText.String()) + } + call, ok := accumCalls[0] + if !ok { + t.Fatal("expected accumCalls[0] to exist") + } + if call.ID != "c1" || call.Name != "bash" { + t.Fatalf("expected tool call c1/bash, got %+v", call) + } + if call.Arguments != "run" { + t.Fatalf("expected arguments %q, got %q", "run", call.Arguments) + } + + // 验证累积状态可用于 buildAssistantMsg + msg := p.buildAssistantMsg(accumText, accumCalls) + if msg.Role != domain.RoleAssistant { + t.Fatalf("expected role assistant, got %q", msg.Role) + } + if !strings.Contains(msg.Content, "partial ") { + t.Fatalf("expected assistant content to contain 'partial', got %q", msg.Content) + } + if len(msg.ToolCalls) != 1 || msg.ToolCalls[0].ID != "c1" { + t.Fatalf("expected assistant tool calls to contain c1, got %+v", msg.ToolCalls) + } +} + +// --- 辅助方法测试 --- + +func TestBuildAssistantMsg_TextOnly(t *testing.T) { + t.Parallel() + + p, _ := New(resolvedConfig("", "")) + var accumText strings.Builder + accumText.WriteString("hello world") + + msg := p.buildAssistantMsg(&accumText, nil) + if msg.Role != domain.RoleAssistant { + t.Fatalf("expected role assistant, got %q", msg.Role) + } + if msg.Content != "hello world" { + t.Fatalf("expected content %q, got %q", "hello world", msg.Content) + } + if len(msg.ToolCalls) != 0 { + t.Fatalf("expected no tool calls, got %+v", msg.ToolCalls) + } +} + +func TestBuildAssistantMsg_WithToolCalls(t *testing.T) { + t.Parallel() + + p, _ := New(resolvedConfig("", "")) + var accumText strings.Builder + accumText.WriteString("done") + accumCalls := map[int]*domain.ToolCall{ + 0: {ID: "call_1", Name: "edit", Arguments: `{"path":"f.go"}`}, + 1: {ID: "call_2", Name: "read", Arguments: `{"path":"f.go"}`}, + } + + msg := p.buildAssistantMsg(&accumText, accumCalls) + if msg.Content != "done" { + t.Fatalf("content mismatch") + } + if len(msg.ToolCalls) != 2 { + t.Fatalf("expected 2 tool calls, got %d", len(msg.ToolCalls)) + } + if msg.ToolCalls[0].Name != "edit" || msg.ToolCalls[1].Name != "read" { + t.Fatalf("unexpected tool calls: %+v", msg.ToolCalls) + } +} + +func TestBuildAssistantMsg_EmptyAccum(t *testing.T) { + t.Parallel() + + p, _ := New(resolvedConfig("", "")) + var accumText strings.Builder + + msg := p.buildAssistantMsg(&accumText, nil) + if msg.Content != "" { + t.Fatalf("expected empty content, got %q", msg.Content) + } + if msg.ToolCalls != nil { + t.Fatal("expected nil ToolCalls when accum is nil") + } +} + +func TestMergeToolCallDeltaWithAccum_SyncsExternalState(t *testing.T) { + t.Parallel() + + events := make(chan domain.StreamEvent, 4) + accumCalls := make(map[int]*domain.ToolCall) + + delta1 := toolCallDelta{ + Index: 0, + ID: "call_acc", + Function: openAIFunctionCall{ + Name: "bash", + Arguments: `{"cmd":"ls"`, + }, + } + if err := mergeToolCallDeltaWithAccum(context.Background(), events, &accumCalls, delta1); err != nil { + t.Fatalf("first delta error = %v", err) + } + + delta2 := toolCallDelta{ + Index: 0, + Function: openAIFunctionCall{ + Arguments: `"}`, + }, + } + if err := mergeToolCallDeltaWithAccum(context.Background(), events, &accumCalls, delta2); err != nil { + t.Fatalf("second delta error = %v", err) + } + + // 验证外部 accumCalls 状态已同步 + call, ok := accumCalls[0] + if !ok { + t.Fatal("expected accumCalls[0] to exist") + } + if call.ID != "call_acc" || call.Name != "bash" { + t.Fatalf("unexpected call state: %+v", call) + } + if call.Arguments != `{"cmd":"ls""}` { + t.Fatalf("expected arguments %q, got %q", `{"cmd":"ls""}`, call.Arguments) + } +} + +func TestMergeToolCallDeltaWithAccum_NilMapInitializes(t *testing.T) { + t.Parallel() + + var accumCalls map[int]*domain.ToolCall // nil map（非指针） + + delta := toolCallDelta{ + Index: 2, + ID: "call_nil", + Function: openAIFunctionCall{ + Name: "read", + }, + } + events := make(chan domain.StreamEvent, 2) + if err := mergeToolCallDeltaWithAccum(context.Background(), events, &accumCalls, delta); err != nil { + t.Fatalf("error = %v", err) + } + + if accumCalls == nil { + t.Fatal("expected accumCalls to be initialized from nil") + } + if accumCalls[2] == nil || accumCalls[2].Name != "read" { + t.Fatalf("unexpected accumCalls[2]: %+v", accumCalls[2]) + } +} + +// --- consumeStream 错误包装测试 --- + +func TestConsumeStream_WrapsNonEOFAsInterrupted(t *testing.T) { + t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") + + p, err := New(resolvedConfig("", "")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + // 使用一个会触发读取错误的 source（模拟网络断开） + errReader := &errReader{err: io.ErrClosedPipe} + accumText := &strings.Builder{} + accums := make(map[int]*domain.ToolCall) + + err = p.consumeStream(context.Background(), errReader, make(chan domain.StreamEvent, 1), accumText, &accums) + if err == nil { + t.Fatal("expected error for broken reader") + } + if !errors.Is(err, domain.ErrStreamInterrupted) { + t.Fatalf("expected ErrStreamInterrupted wrapping, got: %v", err) + } +} + +// TestConsumeStream_FlushesPendingDataOnNonEOFError 验证非 EOF 读取错误发生前， +// 已缓冲但尚未刷新的 data: 行仍会被处理（不会因中断而丢失）。 +func TestConsumeStream_FlushesPendingDataOnNonEOFError(t *testing.T) { + t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") + + p, err := New(resolvedConfig("", "")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + // 构造一段 SSE 数据：包含一个有效 data 行，但紧跟一个错误而非空行。 + // 这模拟了流中断前最后一帧数据还没来得及被空行触发刷新的场景。 + sseData := `data: {"id":"a","object":"chat.completion.chunk","choices":[{"delta":{"content":"hello"},"finish_reason":""}]} +` // 注意：这里有换行符，但由于紧接着是 error，不会被空行刷新 + body := io.MultiReader(strings.NewReader(sseData), &errReader{err: io.ErrClosedPipe}) + + events := make(chan domain.StreamEvent, 10) + accumText := &strings.Builder{} + accums := make(map[int]*domain.ToolCall) + + err = p.consumeStream(context.Background(), body, events, accumText, &accums) + if err == nil { + t.Fatal("expected error for broken reader") + } + if !errors.Is(err, domain.ErrStreamInterrupted) { + t.Fatalf("expected ErrStreamInterrupted, got: %v", err) + } + + // 关键断言：中断前的 data 行必须已被刷新处理，文本累积不为空。 + if accumText.String() != "hello" { + t.Fatalf("expected accumText 'hello', got %q", accumText.String()) + } +} + +// errReader 是一个每次 ReadLine 都返回指定错误的测试辅助类型。 +type errReader struct { + err error +} + +func (e *errReader) Read(p []byte) (int, error) { + return 0, e.err +} + +// roundTripperFunc 将函数适配为 http.RoundTripper 接口，用于测试中 mock HTTP 行为。 +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} diff --git a/internal/provider/openai/sse_reader_test.go b/internal/provider/openai/sse_reader_test.go new file mode 100644 index 00000000..f0d8093b --- /dev/null +++ b/internal/provider/openai/sse_reader_test.go @@ -0,0 +1,193 @@ +package openai + +import ( + "errors" + "io" + "strings" + "testing" + + "neo-code/internal/provider" +) + +// --- boundedSSEReader 单元测试 --- + +func TestBoundedSSEReader_ReadLine_Normal(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + input string + want string + isEOF bool + }{ + { + name: "single line with newline", + input: "data: hello\n", + want: "data: hello", + isEOF: false, + }, + { + name: "line with CRLF", + input: "data: world\r\n", + want: "data: world", + isEOF: false, + }, + { + name: "empty line", + input: "\n", + want: "", + isEOF: false, + }, + { + name: "SSE comment line", + input: ": heartbeat\n", + want: ": heartbeat", + isEOF: false, + }, + { + name: "EOF without trailing newline (io.EOF)", + input: "data: partial", + want: "data: partial", + isEOF: true, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + r := newBoundedSSEReader(strings.NewReader(tt.input)) + got, err := r.ReadLine() + if got != tt.want { + t.Fatalf("ReadLine() = %q, want %q", got, tt.want) + } + if tt.isEOF && !errors.Is(err, io.EOF) { + t.Fatalf("expected io.EOF, got %v", err) + } + if !tt.isEOF && err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} + +func TestBoundedSSEReader_ReadLine_MultipleLines(t *testing.T) { + t.Parallel() + + r := newBoundedSSEReader(strings.NewReader("line1\nline2\n\nline4\n")) + + line1, err := r.ReadLine() + if err != nil || line1 != "line1" { + t.Fatalf("first line: got %q, err = %v", line1, err) + } + + line2, err := r.ReadLine() + if err != nil || line2 != "line2" { + t.Fatalf("second line: got %q, err = %v", line2, err) + } + + // 空行 + empty, err := r.ReadLine() + if err != nil || empty != "" { + t.Fatalf("empty line: got %q, err = %v", empty, err) + } + + line4, err := r.ReadLine() + if err != nil || line4 != "line4" { + t.Fatalf("fourth line: got %q, err = %v", line4, err) + } + + // EOF + _, err = r.ReadLine() + if !errors.Is(err, io.EOF) { + t.Fatalf("expected EOF after all lines, got %v", err) + } +} + +func TestBoundedSSEReader_L1_LineTooLong(t *testing.T) { + t.Parallel() + + longLine := strings.Repeat("x", maxSSELineSize+1) + "\n" + r := newBoundedSSEReader(strings.NewReader(longLine)) + + _, err := r.ReadLine() + if err == nil { + t.Fatal("expected ErrLineTooLong for oversized line") + } + if !errors.Is(err, provider.ErrLineTooLong) { + t.Fatalf("expected ErrLineTooLong, got %v", err) + } +} + +func TestBoundedSSEReader_L1_BoundaryExactLimit(t *testing.T) { + t.Parallel() + + // 恰好等于上限的行应该正常通过（不含 \n） + exactLine := strings.Repeat("a", maxSSELineSize) + "\n" + r := newBoundedSSEReader(strings.NewReader(exactLine)) + + got, err := r.ReadLine() + if err != nil { + t.Fatalf("unexpected error at exact limit: %v", err) + } + if len(got) != maxSSELineSize { + t.Fatalf("expected line length %d, got %d", maxSSELineSize, len(got)) + } +} + +func TestBoundedSSEReader_L3_StreamTooLarge(t *testing.T) { + t.Parallel() + + // 构造输入：多行小内容但总量超过 maxStreamTotalSize + var sb strings.Builder + for i := 0; i < 100; i++ { + sb.WriteString("data: chunk\n") + } + // 填充到超过总量上限 + remaining := maxStreamTotalSize - int64(sb.Len()) + 1 + sb.WriteString(strings.Repeat("x", int(remaining)) + "\n") + + r := newBoundedSSEReader(strings.NewReader(sb.String())) + + // 前面的行应能正常读取 + for i := 0; i < 100; i++ { + _, err := r.ReadLine() + if err != nil { + t.Fatalf("unexpected error on normal line %d: %v", i, err) + } + } + + // 超限的行应返回 ErrStreamTooLarge + _, err := r.ReadLine() + if err == nil { + t.Fatal("expected ErrStreamTooLarge") + } + if !errors.Is(err, provider.ErrStreamTooLarge) { + t.Fatalf("expected ErrStreamTooLarge, got %v", err) + } +} + +func TestTrimLineEnding(t *testing.T) { + t.Parallel() + + tests := []struct { + input string + want string + }{ + {"hello\n", "hello"}, + {"hello\r\n", "hello"}, + {"hello\r\n\n", "hello"}, // 连续换行符全部去除 + {"hello", "hello"}, + {"\n", ""}, + {"\r\n", ""}, + {"\r", ""}, // 孤立 \r 也去除 + {"", ""}, + } + + for _, tt := range tests { + got := trimLineEnding(tt.input) + if got != tt.want { + t.Fatalf("trimLineEnding(%q) = %q, want %q", tt.input, got, tt.want) + } + } +} From aefa4fab7c74975ef832babd239d5c0f7ff98f12 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Mon, 6 Apr 2026 19:44:46 +0800 Subject: [PATCH 3/4] =?UTF-8?q?fix(provider):=E4=BF=AE=E5=A4=8DSSE=20Reade?= =?UTF-8?q?r=20=E7=BC=93=E5=86=B2=E5=8C=BA=E6=BA=A2=E5=87=BA=E4=BB=A5?= =?UTF-8?q?=E5=8F=8A=E9=87=8D=E8=BF=9E=E6=B6=88=E6=81=AF=E6=B1=A1=E6=9F=93?= =?UTF-8?q?=E9=A3=8E=E9=99=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/openai/openai.go | 17 +- internal/provider/openai/openai_test.go | 222 +++++++++++++++++++- internal/provider/openai/sse_reader.go | 30 ++- internal/provider/openai/sse_reader_test.go | 91 +++++++- 4 files changed, 337 insertions(+), 23 deletions(-) diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index 2479de5f..c71e1cca 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -109,6 +109,10 @@ func (p *Provider) DiscoverModels(ctx context.Context) ([]config.ModelDescriptor func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { const maxReconnects = 3 + // 保存原始消息列表的副本，避免重连时反复 append 到同一个切片导致上下文污染 + originalMessages := make([]provider.Message, len(req.Messages)) + copy(originalMessages, req.Messages) + // 跨重连周期持久化的累积状态：已收到的文本和 tool call var ( accumText strings.Builder @@ -117,9 +121,16 @@ func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events ch for attempt := 0; attempt <= maxReconnects; attempt++ { if attempt > 0 { - // 将已累积内容作为 assistant 消息注入，使新请求能从断点继续 - req.Messages = append(req.Messages, - p.buildAssistantMsg(&accumText, accumCalls)) + // 从原始消息出发构造本次请求的完整消息列表 + req.Messages = make([]provider.Message, len(originalMessages), len(originalMessages)+1) + copy(req.Messages, originalMessages) + + // 仅在有实际累积内容时注入 assistant 快照，避免插入空消息 + if accumText.Len() > 0 || len(accumCalls) > 0 { + req.Messages = append(req.Messages, + p.buildAssistantMsg(&accumText, accumCalls)) + } + // 指数退避等待 backoff := time.Duration(1< maxStreamTotalSize { - return "", provider.ErrStreamTooLarge + // L1: 缓冲区溢出 → 单行超过 maxSSELineSize（触发在读取过程中，而非读完后） + if errors.Is(err, bufio.ErrBufferFull) { + return "", provider.ErrLineTooLong } if err != nil && !errors.Is(err, io.EOF) { return "", err } - // L1: 单行长度检查（不含末尾 \n） + // L1 兜底：行内容长度检查（不含末尾 \n） rawLen := len(line) if rawLen > 0 && line[rawLen-1] == '\n' { rawLen-- @@ -55,8 +61,14 @@ func (r *boundedSSEReader) ReadLine() (string, error) { return "", provider.ErrLineTooLong } - // 去除尾部 \r\n - return trimLineEnding(line), err + // L3: 总量检查 + r.totalRead += int64(len(line)) + if r.totalRead > maxStreamTotalSize { + return "", provider.ErrStreamTooLarge + } + + // 将 []byte 转为 string（ReadSlice 返回的底层数据在下次读取时会被覆盖） + return trimLineEnding(string(line)), err } // trimLineEnding 移除行尾的 \r\n 或 \n。 diff --git a/internal/provider/openai/sse_reader_test.go b/internal/provider/openai/sse_reader_test.go index f0d8093b..eb7a1803 100644 --- a/internal/provider/openai/sse_reader_test.go +++ b/internal/provider/openai/sse_reader_test.go @@ -138,26 +138,28 @@ func TestBoundedSSEReader_L1_BoundaryExactLimit(t *testing.T) { func TestBoundedSSEReader_L3_StreamTooLarge(t *testing.T) { t.Parallel() - // 构造输入：多行小内容但总量超过 maxStreamTotalSize + // 构造输入：每行 1KB（远小于 maxSSELineSize），行数足够多使总量超过 maxStreamTotalSize + line := strings.Repeat("x", 1024) + "\n" // 1025 bytes per line + lineSize := int64(len(line)) + var sb strings.Builder - for i := 0; i < 100; i++ { - sb.WriteString("data: chunk\n") + linesToWrite := int(maxStreamTotalSize/lineSize) + 1 + for range linesToWrite { + sb.WriteString(line) } - // 填充到超过总量上限 - remaining := maxStreamTotalSize - int64(sb.Len()) + 1 - sb.WriteString(strings.Repeat("x", int(remaining)) + "\n") r := newBoundedSSEReader(strings.NewReader(sb.String())) // 前面的行应能正常读取 - for i := 0; i < 100; i++ { + expectedNormal := int(maxStreamTotalSize / lineSize) + for range expectedNormal { _, err := r.ReadLine() if err != nil { - t.Fatalf("unexpected error on normal line %d: %v", i, err) + t.Fatalf("unexpected error on normal line: %v", err) } } - // 超限的行应返回 ErrStreamTooLarge + // 超限的行应返回 ErrStreamTooLarge（而非 ErrLineTooLong） _, err := r.ReadLine() if err == nil { t.Fatal("expected ErrStreamTooLarge") @@ -191,3 +193,74 @@ func TestTrimLineEnding(t *testing.T) { } } } + +// TestBoundedSSEReader_L1_NoNewlineEOFAtLimit 验证恰好等于缓冲区大小 +// 但以 EOF 结尾（无 \n）的行能被正常读取并返回 io.EOF。 +func TestBoundedSSEReader_L1_NoNewlineEOFAtLimit(t *testing.T) { + t.Parallel() + + // 不含 \n 的行，长度恰好等于 maxSSELineSize，以 EOF 结尾 + exactLine := strings.Repeat("b", maxSSELineSize) + r := newBoundedSSEReader(strings.NewReader(exactLine)) + + got, err := r.ReadLine() + if err == nil || !errors.Is(err, io.EOF) { + t.Fatalf("expected io.EOF for line without trailing newline, got err=%v", err) + } + if len(got) != maxSSELineSize { + t.Fatalf("expected line length %d, got %d", maxSSELineSize, len(got)) + } +} + +// TestBoundedSSEReader_L1_NoNewlineEOFExceedsLimit 验证超过缓冲区大小 +// 且以 EOF 结尾（无 \n）的行返回 ErrLineTooLong。 +func TestBoundedSSEReader_L1_NoNewlineEOFExceedsLimit(t *testing.T) { + t.Parallel() + + // 不含 \n，长度超过 maxSSELineSize + longLine := strings.Repeat("c", maxSSELineSize+1) + r := newBoundedSSEReader(strings.NewReader(longLine)) + + _, err := r.ReadLine() + if err == nil { + t.Fatal("expected ErrLineTooLong for oversized line without newline") + } + if !errors.Is(err, provider.ErrLineTooLong) { + t.Fatalf("expected ErrLineTooLong, got %v", err) + } +} + +// TestBoundedSSEReader_UnderlyingErrorPropagation 验证底层 reader 的非 EOF 错误 +// 会被正确传播，而不是被 L1/L3 吞掉。 +func TestBoundedSSEReader_UnderlyingErrorPropagation(t *testing.T) { + t.Parallel() + + r := newBoundedSSEReader(&errReader{err: io.ErrClosedPipe}) + _, err := r.ReadLine() + if err == nil { + t.Fatal("expected error from broken reader") + } + if !errors.Is(err, io.ErrClosedPipe) { + t.Fatalf("expected io.ErrClosedPipe, got %v", err) + } +} + +// TestBoundedSSEReader_L1_ThenNormalRead 验证 L1 触发后 reader 状态仍然一致， +// 后续正常行仍可继续读取（如果调用方选择恢复）。 +func TestBoundedSSEReader_L1_ThenNormalRead(t *testing.T) { + t.Parallel() + + // 第一行超长触发 L1，第二行正常 + input := strings.Repeat("x", maxSSELineSize+10) + "\nnormal line\n" + r := newBoundedSSEReader(strings.NewReader(input)) + + // 第一行应返回 ErrLineTooLong + _, err := r.ReadLine() + if !errors.Is(err, provider.ErrLineTooLong) { + t.Fatalf("first line: expected ErrLineTooLong, got %v", err) + } + + // 注意：L1 触发后 bufio.Reader 内部可能已消耗了部分后续数据（缓冲区残留）， + // 因此此测试仅验证 L1 错误被正确返回，不要求后续行一定可读。 + // 这符合实际使用场景——L1 触发后调用方会终止流消费。 +} From 8aadd974fa38ccdc51b0e325f452c8ee3219cde4 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Mon, 6 Apr 2026 20:07:31 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix(provider):=E4=BF=AE=E5=A4=8D=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E6=AC=A1=E6=95=B0=E8=BF=87=E5=A4=9A=E4=BB=A5=E5=8F=8A?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E4=B8=A2=E5=A4=B1=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/errors.go | 20 ++++++ internal/provider/errors_test.go | 86 +++++++++++++++++++++++++ internal/provider/openai/openai.go | 11 ++-- internal/provider/openai/openai_test.go | 8 +++ 4 files changed, 120 insertions(+), 5 deletions(-) diff --git a/internal/provider/errors.go b/internal/provider/errors.go index 29697cca..a2fee38f 100644 --- a/internal/provider/errors.go +++ b/internal/provider/errors.go @@ -108,6 +108,26 @@ func NewTimeoutProviderError(message string) *ProviderError { } } +// MarkNonRetryable 将错误标记为不可重试，用于防止上层重试叠加放大。 +// +// 若错误链中包含 *ProviderError，返回其 Retryable=false 的副本； +// 否则将原始错误包装为 *ProviderError{Code: ErrorCodeUnknown, Retryable: false}。 +// 原始错误通过 Unwrap 保留，不影响 errors.Is/As 对原始哨兵的匹配。 +func MarkNonRetryable(err error) error { + var pErr *ProviderError + if errors.As(err, &pErr) { + clone := *pErr + clone.Retryable = false + return &clone + } + return &ProviderError{ + StatusCode: 0, + Code: ErrorCodeUnknown, + Message: err.Error(), + Retryable: false, + } +} + // IsRecoverableStreamError 判断流读取错误是否可通过透明重连恢复。 // // 不可恢复的情况： diff --git a/internal/provider/errors_test.go b/internal/provider/errors_test.go index 38e6b8c6..eed23978 100644 --- a/internal/provider/errors_test.go +++ b/internal/provider/errors_test.go @@ -301,3 +301,89 @@ func TestIsRecoverableStreamError_UnknownError_NotRecoverable(t *testing.T) { }) } } + +// --- MarkNonRetryable 测试 --- + +func TestMarkNonRetryable_ProviderError(t *testing.T) { + t.Parallel() + + // Retryable=true 的 ProviderError → Retryable=false + retryable := NewProviderErrorFromStatus(http.StatusInternalServerError, "internal") + if !retryable.Retryable { + t.Fatal("setup: expected retryable") + } + + marked := MarkNonRetryable(retryable) + var pErr *ProviderError + if !errors.As(marked, &pErr) { + t.Fatal("marked error should be *ProviderError") + } + if pErr.Retryable { + t.Fatal("MarkNonRetryable should set Retryable=false") + } + if pErr.StatusCode != 500 || pErr.Code != ErrorCodeServer { + t.Fatalf("MarkNonRetryable should preserve StatusCode and Code, got status=%d code=%s", pErr.StatusCode, pErr.Code) + } + + // 原始对象不受影响 + if !retryable.Retryable { + t.Fatal("original ProviderError should not be mutated") + } +} + +func TestMarkNonRetryable_WrappedProviderError(t *testing.T) { + t.Parallel() + + inner := NewProviderErrorFromStatus(http.StatusTooManyRequests, "rate limited") + wrapped := fmt.Errorf("layer: %w", inner) + + marked := MarkNonRetryable(wrapped) + var pErr *ProviderError + if !errors.As(marked, &pErr) { + t.Fatal("marked error should contain *ProviderError") + } + if pErr.Retryable { + t.Fatal("MarkNonRetryable on wrapped should set Retryable=false") + } +} + +func TestMarkNonRetryable_NonProviderError(t *testing.T) { + t.Parallel() + + // 非 ProviderError → 包装为 ProviderError{Retryable: false} + generic := errors.New("some error") + marked := MarkNonRetryable(generic) + + var pErr *ProviderError + if !errors.As(marked, &pErr) { + t.Fatal("marked error should be *ProviderError") + } + if pErr.Retryable { + t.Fatal("should be non-retryable") + } + if pErr.Code != ErrorCodeUnknown { + t.Fatalf("expected code %s, got %s", ErrorCodeUnknown, pErr.Code) + } + if !strings.Contains(pErr.Message, "some error") { + t.Fatalf("message should contain original error text, got: %s", pErr.Message) + } +} + +func TestMarkNonRetryable_AlreadyNonRetryable(t *testing.T) { + t.Parallel() + + // 已经 Retryable=false 的 ProviderError → 保持 false + nonRetryable := NewProviderErrorFromStatus(http.StatusUnauthorized, "bad key") + if nonRetryable.Retryable { + t.Fatal("setup: 401 should be non-retryable") + } + + marked := MarkNonRetryable(nonRetryable) + var pErr *ProviderError + if !errors.As(marked, &pErr) { + t.Fatal("marked error should be *ProviderError") + } + if pErr.Retryable { + t.Fatal("should still be non-retryable") + } +} diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index c71e1cca..6e3a59c7 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -144,9 +144,13 @@ func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events ch if err == nil { return nil } - if !provider.IsRecoverableStreamError(err) || attempt == maxReconnects { + if !provider.IsRecoverableStreamError(err) { return err } + // 可恢复但重连次数已耗尽 → 标记为不可重试，防止上层 runtime 重试叠加放大。 + if attempt == maxReconnects { + return provider.MarkNonRetryable(err) + } } return nil // unreachable，但满足编译器 } @@ -318,7 +322,7 @@ func (p *Provider) consumeStream( if flushErr := flushPendingData(); flushErr != nil { return flushErr } - return fmt.Errorf("%w: %v", provider.ErrStreamInterrupted, err) + return fmt.Errorf("%w: %w", provider.ErrStreamInterrupted, err) } trimmed := line @@ -415,9 +419,6 @@ func emitToolCallDelta(ctx context.Context, events chan<- provider.StreamEvent, // emitMessageDone 发送消息完成事件。 func emitMessageDone(ctx context.Context, events chan<- provider.StreamEvent, finishReason string, usage *provider.Usage) error { - if events == nil { - return nil - } return emitStreamEvent(ctx, events, provider.NewMessageDoneStreamEvent(finishReason, usage)) } diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index db61452d..9085e54b 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -1146,6 +1146,14 @@ func TestProviderChatReconnect_MaxRetriesExhausted(t *testing.T) { if err == nil { t.Fatal("expected error after exhausting retries") } + // 重连耗尽后，错误应被标记为不可重试，防止上层 runtime 再次重试叠加放大。 + var pErr *domain.ProviderError + if !errors.As(err, &pErr) { + t.Fatalf("expected *ProviderError after retry exhaustion, got: %T: %v", err, err) + } + if pErr.Retryable { + t.Fatal("error should be non-retryable after reconnect exhaustion") + } // 初始1次 + 最大3次重连 = 最多4次尝试 if attempt > 4 { t.Fatalf("too many attempts: %d (max should be 4)", attempt)