From daec7444ef0badb527ffd73c57ec553da330b292 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Mon, 6 Apr 2026 23:08:30 +0800 Subject: [PATCH] =?UTF-8?q?pref(provider)=EF=BC=9A=E5=88=A0=E6=8E=89?= =?UTF-8?q?=E9=80=8F=E6=98=8E=E9=87=8D=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/errors.go | 62 --- internal/provider/errors_test.go | 217 ---------- internal/provider/openai/openai.go | 117 +---- internal/provider/openai/openai_test.go | 553 +----------------------- 4 files changed, 19 insertions(+), 930 deletions(-) diff --git a/internal/provider/errors.go b/internal/provider/errors.go index a2fee38f..daff0da6 100644 --- a/internal/provider/errors.go +++ b/internal/provider/errors.go @@ -1,10 +1,8 @@ package provider import ( - "context" "errors" "fmt" - "net" "net/http" ) @@ -107,63 +105,3 @@ func NewTimeoutProviderError(message string) *ProviderError { Retryable: true, // 超时默认可重试 } } - -// 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 判断流读取错误是否可通过透明重连恢复。 -// -// 不可恢复的情况： -// - 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/errors_test.go b/internal/provider/errors_test.go index eed23978..e6e9ac4d 100644 --- a/internal/provider/errors_test.go +++ b/internal/provider/errors_test.go @@ -1,10 +1,8 @@ package provider import ( - "context" "errors" "fmt" - "io" "net/http" "strings" "testing" @@ -172,218 +170,3 @@ 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) - } - }) - } -} - -// --- 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 6e3a59c7..2e2cbaf8 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -101,69 +101,9 @@ func (p *Provider) DiscoverModels(ctx context.Context) ([]config.ModelDescriptor return config.MergeModelDescriptors(descriptors), nil } -// Chat 发起 SSE 流式对话请求，支持透明重连。 -// -// 流中途断连时，将已累积的 assistant 消息（文本 + tool call）注入请求上下文， -// 利用 OpenAI 多轮对话语义实现断点续传，对上层调用方透明。 -// 最多重连 maxReconnects 次；不可恢复错误直接返回。 +// Chat 发起 SSE 流式对话请求。 +// 流中途断连或协议错误时直接返回错误，由上层调用方决定重试策略。 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 - accumCalls map[int]*provider.ToolCall - ) - - for attempt := 0; attempt <= maxReconnects; attempt++ { - if attempt > 0 { - // 从原始消息出发构造本次请求的完整消息列表 - 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< 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 9085e54b..bfbca884 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -1,11 +1,9 @@ package openai import ( - "bytes" "context" "encoding/json" "errors" - "fmt" "io" "net/http" "net/http/httptest" @@ -618,7 +616,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), &strings.Builder{}, new(map[int]*domain.ToolCall)) + err = provider.consumeStream(context.Background(), strings.NewReader("data: {not-json}\n\n"), make(chan domain.StreamEvent, 1)) if err == nil || !strings.Contains(err.Error(), "decode stream chunk") { t.Fatalf("expected dirty JSON decode error, got %v", err) } @@ -1032,318 +1030,8 @@ func TestProviderChatEmitsFullEventStream(t *testing.T) { } } -// --- 透明重连测试 --- - -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") - } - // 重连耗尽后，错误应被标记为不可重试，防止上层 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) - } -} - -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)) - } - // 使用 map 检查，避免依赖 Go map 迭代的不确定顺序 - names := make(map[string]bool) - for _, tc := range msg.ToolCalls { - names[tc.Name] = true - } - if !names["edit"] || !names["read"] { - t.Fatalf("expected tool calls 'edit' and 'read', got %+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) { @@ -1356,10 +1044,8 @@ func TestConsumeStream_WrapsNonEOFAsInterrupted(t *testing.T) { // 使用一个会触发读取错误的 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) + err = p.consumeStream(context.Background(), errReader, make(chan domain.StreamEvent, 1)) if err == nil { t.Fatal("expected error for broken reader") } @@ -1385,10 +1071,8 @@ func TestConsumeStream_FlushesPendingDataOnNonEOFError(t *testing.T) { 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) + err = p.consumeStream(context.Background(), body, events) if err == nil { t.Fatal("expected error for broken reader") } @@ -1396,9 +1080,16 @@ func TestConsumeStream_FlushesPendingDataOnNonEOFError(t *testing.T) { t.Fatalf("expected ErrStreamInterrupted, got: %v", err) } - // 关键断言：中断前的 data 行必须已被刷新处理，文本累积不为空。 - if accumText.String() != "hello" { - t.Fatalf("expected accumText 'hello', got %q", accumText.String()) + // 关键断言：中断前的 data 行必须已被刷新处理，应有 text_delta 事件。 + 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 from flushed pending data") } } @@ -1410,221 +1101,3 @@ type errReader struct { 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) -} - -// --- 重连消息完整性测试 --- - -// TestReconnect_NoEmptyAssistantOnFirstFailure 验证首次请求失败（未收到任何 SSE 数据） -// 时，重连不会向消息列表注入空的 assistant 消息。 -func TestReconnect_NoEmptyAssistantOnFirstFailure(t *testing.T) { - t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") - - attempt := 0 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - attempt++ - - // 解码请求中的消息，验证消息结构 - var payload chatCompletionRequest - if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { - t.Errorf("decode request: %v", err) - w.WriteHeader(http.StatusBadRequest) - return - } - - if attempt == 1 { - // 首次请求：返回 500（无 SSE 数据），不应注入空 assistant 消息 - if len(payload.Messages) != 1 { - t.Errorf("first attempt: expected 1 message, got %d", len(payload.Messages)) - } - w.WriteHeader(http.StatusInternalServerError) - _, _ = w.Write([]byte(`{"error":{"message":"temporarily unavailable"}}`)) - return - } - - // 第二次请求：仍应只有原始 1 条消息（无空 assistant 注入） - if len(payload.Messages) != 1 { - t.Errorf("second attempt: expected 1 message (no empty assistant), got %d; messages: %+v", - len(payload.Messages), payload.Messages) - } - for _, msg := range payload.Messages { - if msg.Role == "assistant" { - t.Errorf("second attempt: unexpected assistant message injected: %+v", msg) - } - } - - w.Header().Set("Content-Type", "text/event-stream") - writeSSEChunk(t, w, map[string]any{ - "choices": []map[string]any{ - {"index": 0, "delta": map[string]any{"content": "ok"}}, - }, - }) - _, _ = 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, 4) - 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) - } - if attempt != 2 { - t.Fatalf("expected exactly 2 attempts, got %d", attempt) - } -} - -// TestReconnect_SingleAssistantSnapshotNotDuplicated 验证多次重连时，每次请求 -// 只包含原始消息 + 恰好 1 条 assistant 快照，不会出现旧快照残留。 -// 使用自定义 RoundTripper 模拟流中途中断（非 EOF 错误）。 -func TestReconnect_SingleAssistantSnapshotNotDuplicated(t *testing.T) { - t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key") - - attempt := 0 - - // 使用 httptest.Server 作为成功响应的代理，RoundTripper 控制中断 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "text/event-stream") - writeSSEChunk(t, w, map[string]any{ - "choices": []map[string]any{ - {"index": 0, "delta": map[string]any{"content": " done"}}, - }, - }) - _, _ = w.Write([]byte("data: [DONE]\n\n")) - })) - defer server.Close() - - rt := roundTripperFunc(func(req *http.Request) (*http.Response, error) { - attempt++ - - // 读取并缓存请求体，以便解码后仍可转发给真实 transport - bodyBytes, err := io.ReadAll(req.Body) - _ = req.Body.Close() - if err != nil { - return nil, fmt.Errorf("read request body: %w", err) - } - - // 解码请求消息，用于断言 - var payload chatCompletionRequest - if err := json.Unmarshal(bodyBytes, &payload); err != nil { - return nil, fmt.Errorf("decode request: %w", err) - } - - // 恢复请求体，确保后续 transport 可以读取 - req.Body = io.NopCloser(bytes.NewReader(bodyBytes)) - req.ContentLength = int64(len(bodyBytes)) - - switch attempt { - case 1: - // 首次请求：正常消息（system + user），发送部分 SSE 后流中断 - if len(payload.Messages) != 2 { - t.Errorf("attempt 1: expected 2 messages, got %d", len(payload.Messages)) - } - sseData := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"partial\"}}]}\n\n" - body := io.MultiReader(strings.NewReader(sseData), &errReader{err: io.ErrClosedPipe}) - return &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"text/event-stream"}}, - Body: io.NopCloser(body), - }, nil - - case 2: - // 第二次请求：应包含原始 2 条 + 1 条 assistant（"partial"），再次中断 - if len(payload.Messages) != 3 { - t.Errorf("attempt 2: expected 3 messages, got %d; messages: %+v", - len(payload.Messages), payload.Messages) - } - assistMsg := payload.Messages[2] - if assistMsg.Role != "assistant" || assistMsg.Content != "partial" { - t.Errorf("attempt 2: expected assistant content 'partial', got role=%q content=%q", - assistMsg.Role, assistMsg.Content) - } - assistCount := countMessagesByRole(payload.Messages, "assistant") - if assistCount != 1 { - t.Errorf("attempt 2: expected 1 assistant message, got %d", assistCount) - } - - sseData := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\" more\"}}]}\n\n" - body := io.MultiReader(strings.NewReader(sseData), &errReader{err: io.ErrClosedPipe}) - return &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"text/event-stream"}}, - Body: io.NopCloser(body), - }, nil - - default: - // 第三次请求：应包含原始 2 条 + 1 条 assistant（"partial more"） - if len(payload.Messages) != 3 { - t.Errorf("attempt 3: expected 3 messages, got %d; messages: %+v", - len(payload.Messages), payload.Messages) - } - assistMsg := payload.Messages[2] - if assistMsg.Role != "assistant" || assistMsg.Content != "partial more" { - t.Errorf("attempt 3: expected assistant content 'partial more', got role=%q content=%q", - assistMsg.Role, assistMsg.Content) - } - // 确认仍然只有 1 条 assistant 消息（旧快照未残留） - assistCount := countMessagesByRole(payload.Messages, "assistant") - if assistCount != 1 { - t.Errorf("attempt 3: expected 1 assistant message, got %d", assistCount) - } - - // 委托给 test server 返回完整响应 - return http.DefaultTransport.RoundTrip(req) - } - }) - - p, err := New(resolvedConfig(server.URL, config.OpenAIDefaultModel), withTransport(rt)) - if err != nil { - t.Fatalf("New() error = %v", err) - } - - events := make(chan domain.StreamEvent, 16) - err = p.Chat(context.Background(), domain.ChatRequest{ - SystemPrompt: "you are helpful", - Model: config.OpenAIDefaultModel, - Messages: []domain.Message{{Role: "user", Content: "hello"}}, - }, events) - if err != nil { - t.Fatalf("Chat() should succeed after reconnect, got: %v", err) - } - if attempt != 3 { - t.Fatalf("expected exactly 3 attempts, got %d", attempt) - } - - // 验证最终累积的文本（三次请求的增量合并） - var fullText strings.Builder - for _, evt := range drainStreamEvents(events) { - if evt.Type == domain.StreamEventTextDelta { - fullText.WriteString(requireTextDeltaPayload(t, evt).Text) - } - } - expectedText := "partial more done" - if fullText.String() != expectedText { - t.Fatalf("expected full text %q, got %q", expectedText, fullText.String()) - } -} - -// countMessagesByRole 统计消息列表中指定角色的消息数量。 -func countMessagesByRole(messages []openAIMessage, role string) int { - count := 0 - for _, msg := range messages { - if msg.Role == role { - count++ - } - } - return count -}