diff --git a/CHANGELOG.md b/CHANGELOG.md index 6aecc8d..1d28977 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,12 @@ Write each change in both `### English` and `### 中文` under `## Unreleased`. ### English +- Strip empty WorkBuddy chat-stream deltas so clients do not render a flood of blank thinking chunks + ### 中文 +- WorkBuddy 的 chat 流式不再带上空的 `content` / `reasoning_content`,避免客户端刷出空白思考 + ## 0.3.3 - 2026-09-07 ### English diff --git a/internal/providers/workbuddy/client.go b/internal/providers/workbuddy/client.go index 98dbce6..f8703ab 100644 --- a/internal/providers/workbuddy/client.go +++ b/internal/providers/workbuddy/client.go @@ -441,7 +441,7 @@ func (c *Client) ChatStream(ctx context.Context, accountID string, req translate resp.Body.Close() return nil, classifiedError(resp.StatusCode, body) } - return resp, nil + return rewriteChatStream(resp), nil } func outcomeFromAggregate(aggregate map[string]any) (providers.ChatOutcome, error) { diff --git a/internal/providers/workbuddy/client_test.go b/internal/providers/workbuddy/client_test.go index 9a045a9..0da96ba 100644 --- a/internal/providers/workbuddy/client_test.go +++ b/internal/providers/workbuddy/client_test.go @@ -504,6 +504,73 @@ func TestErrorMapping(t *testing.T) { } } +func TestCopySanitizedSSEDropsEmptyThinkingDeltas(t *testing.T) { + input := strings.Join([]string{ + `data: {"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning_content":"think","function_call":null,"refusal":"","tool_calls":[],"extra_fields":null},"logprobs":null,"finish_reason":""}]}`, + `data: {"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"OK","reasoning_content":"","function_call":null,"refusal":"","tool_calls":[],"extra_fields":null},"logprobs":null,"finish_reason":""}]}`, + `data: {"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning_content":"","function_call":{"name":"","arguments":""},"refusal":"","tool_calls":[],"extra_fields":null},"logprobs":null,"finish_reason":"stop"}],"usage":{"prompt_tokens":16,"completion_tokens":2}}`, + `data: [DONE]`, + "", + }, "\n\n") + var out strings.Builder + if err := copySanitizedSSE(strings.NewReader(input), &out); err != nil { + t.Fatal(err) + } + text := out.String() + if strings.Contains(text, `"content":""`) || strings.Contains(text, `"reasoning_content":""`) || strings.Contains(text, `"function_call"`) { + t.Fatalf("empty delta fields leaked: %s", text) + } + if !strings.Contains(text, `"reasoning_content":"think"`) || !strings.Contains(text, `"content":"OK"`) { + t.Fatalf("real deltas dropped: %s", text) + } + if strings.Count(text, `"role":"assistant"`) != 1 { + t.Fatalf("role should appear once, got %s", text) + } + if !strings.Contains(text, `"finish_reason":"stop"`) || !strings.Contains(text, `"prompt_tokens":16`) || !strings.Contains(text, "data: [DONE]") { + t.Fatalf("terminal chunk missing: %s", text) + } +} + +func TestChatStreamStripsEmptyWorkBuddyDeltas(t *testing.T) { + payload, _ := Credential{AccessToken: "at", UID: "u1", Domain: "codebuddy.cn", ExpiresAt: 4102444800}.Encode() + store := &memStore{items: map[string][]byte{"acc1": payload}} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == pathModelsCN { + _ = json.NewEncoder(w).Encode(map[string]any{"code": 0, "data": map[string]any{ + "models": []map[string]any{{"id": "glm-5.3-flash", "name": "GLM"}}, + "agents": []map[string]any{{"name": "cli", "models": []string{"glm-5.3-flash"}}}, + }}) + return + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte( + `data: {"choices":[{"delta":{"role":"assistant","content":"","reasoning_content":"think","function_call":null,"refusal":"","tool_calls":[]},"finish_reason":""}]}` + "\n\n" + + `data: {"choices":[{"delta":{"role":"assistant","content":"OK","reasoning_content":"","function_call":null},"finish_reason":""}]}` + "\n\n" + + `data: {"choices":[{"delta":{"content":"","reasoning_content":"","function_call":{"name":"","arguments":""}},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}` + "\n\n" + + "data: [DONE]\n\n", + )) + })) + defer server.Close() + client := NewClient(store) + client.http = server.Client() + client.http.Transport = rewriteTransport{server: server.URL, round: server.Client().Transport} + resp, err := client.ChatStream(context.Background(), "acc1", translate.ChatRequest{ + Model: "glm-5.3-flash", Messages: []translate.ChatMessage{{Role: "user", Content: "hi"}}, + }) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + text := string(body) + if strings.Contains(text, `"content":""`) || strings.Contains(text, `"reasoning_content":""`) { + t.Fatalf("empty thinking leaked: %s", text) + } + if !strings.Contains(text, `"reasoning_content":"think"`) || !strings.Contains(text, `"content":"OK"`) || !strings.Contains(text, "data: [DONE]") { + t.Fatalf("stream=%s", text) + } +} + func TestPrepareBodyForcesStreamAndStringToolChoice(t *testing.T) { out := PrepareBody([]byte(`{"model":"m","stream":false,"tool_choice":{"type":"function","function":{"name":"get_time"}}}`)) var body map[string]any diff --git a/internal/providers/workbuddy/sse.go b/internal/providers/workbuddy/sse.go index 2c5c447..6592c6b 100644 --- a/internal/providers/workbuddy/sse.go +++ b/internal/providers/workbuddy/sse.go @@ -2,9 +2,11 @@ package workbuddy import ( "bufio" + "bytes" "encoding/json" "fmt" "io" + "net/http" "sort" "strings" "time" @@ -163,3 +165,142 @@ func mergeToolCalls(merged map[int]map[string]any, order *[]int, raw json.RawMes } } } + +// rewriteChatStream strips empty WorkBuddy delta fields. Upstream chat chunks +// always include content:"", reasoning_content:"", refusal:"", tool_calls:[], +// and a dummy function_call; OpenAI-compatible clients then render a flood of +// blank thinking events. +func rewriteChatStream(resp *http.Response) *http.Response { + pr, pw := io.Pipe() + go func() { + defer resp.Body.Close() + defer pw.Close() + if err := copySanitizedSSE(resp.Body, pw); err != nil { + _ = pw.CloseWithError(err) + } + }() + out := *resp + out.Body = pr + out.ContentLength = -1 + header := resp.Header.Clone() + header.Del("Content-Length") + out.Header = header + return &out +} + +func copySanitizedSSE(src io.Reader, dst io.Writer) error { + scanner := bufio.NewScanner(src) + scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) + sentRole := false + for scanner.Scan() { + line := strings.TrimSuffix(scanner.Text(), "\r") + if line == "" { + continue + } + if !strings.HasPrefix(line, "data:") { + if _, err := fmt.Fprintf(dst, "%s\n", line); err != nil { + return err + } + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if payload == "" { + continue + } + cleaned, ok := sanitizeSSEPayload(payload, &sentRole) + if !ok { + continue + } + if _, err := fmt.Fprintf(dst, "data: %s\n\n", cleaned); err != nil { + return err + } + } + return scanner.Err() +} + +func sanitizeSSEPayload(payload string, sentRole *bool) (string, bool) { + if payload == "[DONE]" { + return payload, true + } + decoder := json.NewDecoder(bytes.NewReader([]byte(payload))) + decoder.UseNumber() + var chunk map[string]any + if err := decoder.Decode(&chunk); err != nil { + return payload, true + } + keep := chunk["usage"] != nil + if choices, ok := chunk["choices"].([]any); ok { + for _, raw := range choices { + choice, _ := raw.(map[string]any) + if choice == nil { + continue + } + if delta, ok := choice["delta"].(map[string]any); ok { + sanitizeDelta(delta) + if sentRole != nil { + if _, hasRole := delta["role"]; hasRole && *sentRole { + delete(delta, "role") + } + } + if len(delta) > 0 { + keep = true + if sentRole != nil { + if _, hasRole := delta["role"]; hasRole { + *sentRole = true + } + } + } + } + switch finish := choice["finish_reason"].(type) { + case string: + if finish == "" { + delete(choice, "finish_reason") + } else { + keep = true + } + case nil: + delete(choice, "finish_reason") + } + } + } + if !keep { + return "", false + } + encoded, err := json.Marshal(chunk) + if err != nil { + return payload, true + } + return string(encoded), true +} + +func sanitizeDelta(delta map[string]any) { + dropEmptyString(delta, "content") + dropEmptyString(delta, "reasoning_content") + dropEmptyString(delta, "refusal") + if extra, ok := delta["extra_fields"]; ok && extra == nil { + delete(delta, "extra_fields") + } + if calls, ok := delta["tool_calls"].([]any); ok && len(calls) == 0 { + delete(delta, "tool_calls") + } + if delta["tool_calls"] == nil { + delete(delta, "tool_calls") + } + switch call := delta["function_call"].(type) { + case nil: + delete(delta, "function_call") + case map[string]any: + name, _ := call["name"].(string) + args, _ := call["arguments"].(string) + if name == "" && args == "" { + delete(delta, "function_call") + } + } +} + +func dropEmptyString(delta map[string]any, key string) { + value, ok := delta[key].(string) + if ok && value == "" { + delete(delta, key) + } +}