From 41321251cb8845f661b810d2f0d76be3a5980f80 Mon Sep 17 00:00:00 2001 From: nguyenha935 Date: Thu, 2 Jul 2026 00:54:29 +0700 Subject: [PATCH] fix: infer text-encoded tool calls by schema (#1320) Co-authored-by: GoClaw Operator --- internal/providers/adapter_openai.go | 6 +- .../providers/control_output_normalizer.go | 399 ++++++++++++++++++ .../control_output_normalizer_test.go | 313 ++++++++++++++ internal/providers/openai_chat.go | 20 +- internal/providers/openai_http.go | 29 +- 5 files changed, 757 insertions(+), 10 deletions(-) create mode 100644 internal/providers/control_output_normalizer.go create mode 100644 internal/providers/control_output_normalizer_test.go diff --git a/internal/providers/adapter_openai.go b/internal/providers/adapter_openai.go index 6e744441..d8334dc5 100644 --- a/internal/providers/adapter_openai.go +++ b/internal/providers/adapter_openai.go @@ -112,8 +112,10 @@ func (a *OpenAIAdapter) FromStreamChunk(data []byte) (*StreamChunk, error) { // Text content if delta.Content != "" { - sc.Content = delta.Content - hasContent = true + normalized := normalizeControlOutput(delta.Content, "", nil) + sc.Content = normalized.Content + sc.Thinking = strings.TrimSpace(strings.Join([]string{sc.Thinking, normalized.Thinking}, "\n")) + hasContent = sc.Content != "" || sc.Thinking != "" } if !hasContent { diff --git a/internal/providers/control_output_normalizer.go b/internal/providers/control_output_normalizer.go new file mode 100644 index 00000000..095f8b33 --- /dev/null +++ b/internal/providers/control_output_normalizer.go @@ -0,0 +1,399 @@ +package providers + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "regexp" + "strings" +) + +const ( + kimiToolSectionBegin = "<|tool_calls_section_begin|>" + kimiToolSectionEnd = "<|tool_calls_section_end|>" + kimiToolCallBegin = "<|tool_call_begin|>" + kimiToolArgsBegin = "<|tool_call_argument_begin|>" + kimiToolCallEnd = "<|tool_call_end|>" +) + +type controlOutput struct { + Content string + Thinking string + ToolCalls []ToolCall +} + +type controlOutputNormalizer struct { + tools []ToolDefinition + pending string +} + +func newControlOutputNormalizer(tools []ToolDefinition) *controlOutputNormalizer { + return &controlOutputNormalizer{tools: tools} +} + +func (n *controlOutputNormalizer) Append(delta string) controlOutput { + n.pending += delta + return n.drain(false) +} + +func (n *controlOutputNormalizer) Finish() controlOutput { + return n.drain(true) +} + +func (n *controlOutputNormalizer) drain(final bool) controlOutput { + var out controlOutput + for n.pending != "" { + cut, wait := safeControlCut(n.pending, final) + if wait { + break + } + if cut == 0 { + break + } + part := n.pending[:cut] + n.pending = n.pending[cut:] + normalized := normalizeControlOutput(part, "", n.tools) + out.Content += normalized.Content + out.Thinking = joinThinking(out.Thinking, normalized.Thinking) + out.ToolCalls = append(out.ToolCalls, normalized.ToolCalls...) + } + return out +} + +func normalizeControlOutput(content, thinking string, tools []ToolDefinition) controlOutput { + out := controlOutput{Content: content, Thinking: thinking} + out.Content, out.ToolCalls = extractKimiTextToolCalls(out.Content, tools) + + var extracted string + out.Content, extracted = extractThinkingTags(out.Content) + out.Thinking = joinThinking(out.Thinking, extracted) + + out.Content = strings.TrimSpace(out.Content) + return out +} + +func safeControlCut(text string, final bool) (int, bool) { + if final { + return len(text), false + } + + start := firstControlStart(text) + if start >= 0 { + if start > 0 { + return start, false + } + if end := completeControlEnd(text); end > 0 { + return end, false + } + return 0, true + } + + keep := controlPrefixSuffixLen(text) + if keep > 0 { + if keep == len(text) { + return 0, true + } + return len(text) - keep, false + } + return len(text), false +} + +func firstControlStart(text string) int { + start := strings.Index(text, kimiToolSectionBegin) + lower := strings.ToLower(text) + for _, token := range thinkingStartTokens { + if idx := strings.Index(lower, token); idx >= 0 && (start < 0 || idx < start) { + start = idx + } + } + return start +} + +func completeControlEnd(text string) int { + if strings.HasPrefix(text, kimiToolSectionBegin) { + if idx := strings.Index(text, kimiToolSectionEnd); idx >= 0 { + return idx + len(kimiToolSectionEnd) + } + return 0 + } + + lower := strings.ToLower(text) + for _, tag := range thinkingTags { + open := "<" + tag + if strings.HasPrefix(lower, open) { + closeStart := strings.Index(lower, "") + if closeEnd < 0 { + return 0 + } + return closeStart + closeEnd + 1 + } + } + return 0 +} + +func controlPrefixSuffixLen(text string) int { + lower := strings.ToLower(text) + maxKeep := 0 + for _, token := range controlStartTokens { + candidate := text + if strings.HasPrefix(token, "<|") { + candidate = text + } else { + candidate = lower + } + limit := len(token) - 1 + if len(candidate) < limit { + limit = len(candidate) + } + for n := limit; n > maxKeep; n-- { + if strings.HasPrefix(token, candidate[len(candidate)-n:]) { + maxKeep = n + break + } + } + } + return maxKeep +} + +var ( + thinkingTags = []string{"redacted_thinking", "antthinking", "thinking", "thought", "think"} + thinkingStartTokens = []string{"(.*?)<\|tool_calls_section_end\|>`) + kimiToolCallRe = regexp.MustCompile(`(?s)<\|tool_call_begin\|>(.*?)<\|tool_call_argument_begin\|>(.*?)<\|tool_call_end\|>`) + thinkingTagRes = []*regexp.Regexp{ + regexp.MustCompile(`(?is)]*>(.*?)`), + regexp.MustCompile(`(?is)]*>(.*?)`), + regexp.MustCompile(`(?is)]*>(.*?)`), + regexp.MustCompile(`(?is)]*>(.*?)`), + regexp.MustCompile(`(?is)]*>(.*?)`), + } +) + +func extractKimiTextToolCalls(content string, tools []ToolDefinition) (string, []ToolCall) { + if !strings.Contains(content, kimiToolSectionBegin) { + return content, nil + } + + var calls []ToolCall + cleaned := kimiToolSectionRe.ReplaceAllStringFunc(content, func(section string) string { + matches := kimiToolCallRe.FindAllStringSubmatch(section, -1) + for _, match := range matches { + call, ok := parseKimiTextToolCall(match[1], match[2], tools, len(calls)) + if ok { + calls = append(calls, call) + } + } + return "" + }) + return cleaned, calls +} + +func parseKimiTextToolCall(rawHeader, rawArgs string, tools []ToolDefinition, index int) (ToolCall, bool) { + args := make(map[string]any) + var parseErr string + trimmedArgs := strings.TrimSpace(rawArgs) + if err := json.Unmarshal([]byte(trimmedArgs), &args); err != nil && trimmedArgs != "" { + parseErr = fmt.Sprintf("malformed JSON (%d chars): %v", len(rawArgs), err) + } + + name, ok := resolveTextToolName(rawHeader, args, parseErr == "", tools) + if !ok { + slog.Warn("provider_control: text tool call stripped without deterministic tool match", + "header", strings.TrimSpace(rawHeader), "tools", len(functionToolNames(tools))) + return ToolCall{}, false + } + + if parseErr != "" { + slog.Warn("provider_control: failed to parse text tool call arguments", + "tool", name, "raw_len", len(rawArgs), "error", parseErr) + } + + id := textToolCallID(rawHeader, index) + return ToolCall{ + ID: truncateToolCallID(id), + Name: name, + Arguments: args, + ParseError: parseErr, + }, true +} + +func resolveTextToolName(rawHeader string, args map[string]any, argsValid bool, tools []ToolDefinition) (string, bool) { + header := strings.TrimSpace(rawHeader) + names := functionToolNames(tools) + for _, name := range names { + if header == name { + return name, true + } + } + if len(names) == 1 { + return names[0], true + } + if argsValid { + matches := matchingToolNamesByArgs(args, tools) + if len(matches) == 1 { + return matches[0], true + } + } + return "", false +} + +func functionToolNames(tools []ToolDefinition) []string { + names := make([]string, 0, len(tools)) + for _, tool := range tools { + if tool.Type == "function" && tool.Function != nil && strings.TrimSpace(tool.Function.Name) != "" { + names = append(names, strings.TrimSpace(tool.Function.Name)) + } + } + return names +} + +func matchingToolNamesByArgs(args map[string]any, tools []ToolDefinition) []string { + if len(args) == 0 { + return nil + } + matches := make([]string, 0, 1) + for _, tool := range tools { + if tool.Type != "function" || tool.Function == nil || strings.TrimSpace(tool.Function.Name) == "" { + continue + } + if toolSchemaMatchesArgs(tool.Function.Parameters, args) { + matches = append(matches, strings.TrimSpace(tool.Function.Name)) + } + } + return matches +} + +func toolSchemaMatchesArgs(parameters map[string]any, args map[string]any) bool { + required := schemaStringList(parameters["required"]) + if len(required) == 0 { + return false + } + properties := schemaProperties(parameters["properties"]) + if len(properties) == 0 { + return false + } + for _, key := range required { + if _, ok := args[key]; !ok { + return false + } + } + for key := range args { + if _, ok := properties[key]; !ok { + return false + } + } + return true +} + +func schemaStringList(value any) []string { + switch v := value.(type) { + case []string: + out := make([]string, 0, len(v)) + for _, item := range v { + if item = strings.TrimSpace(item); item != "" { + out = append(out, item) + } + } + return out + case []any: + out := make([]string, 0, len(v)) + for _, item := range v { + if s, ok := item.(string); ok { + if s = strings.TrimSpace(s); s != "" { + out = append(out, s) + } + } + } + return out + default: + return nil + } +} + +func schemaProperties(value any) map[string]struct{} { + props := make(map[string]struct{}) + switch v := value.(type) { + case map[string]any: + for key := range v { + if key = strings.TrimSpace(key); key != "" { + props[key] = struct{}{} + } + } + } + return props +} + +func textToolCallID(rawHeader string, index int) string { + header := strings.TrimSpace(rawHeader) + if strings.HasPrefix(header, "call_") { + return header + } + sum := sha256.Sum256([]byte(fmt.Sprintf("%s:%d", header, index))) + return "call_" + hex.EncodeToString(sum[:])[:16] +} + +func extractThinkingTags(content string) (string, string) { + lower := strings.ToLower(content) + if !strings.Contains(lower, " 1 && strings.TrimSpace(match[1]) != "" { + thinkingParts = append(thinkingParts, strings.TrimSpace(match[1])) + } + } + cleaned = re.ReplaceAllString(cleaned, "") + } + + cleaned, dangling := extractDanglingThinkingTag(cleaned) + if dangling != "" { + thinkingParts = append(thinkingParts, dangling) + } + + return cleaned, strings.Join(thinkingParts, "\n") +} + +func extractDanglingThinkingTag(content string) (string, string) { + lower := strings.ToLower(content) + idx := -1 + for _, token := range thinkingStartTokens { + if found := strings.Index(lower, token); found >= 0 && (idx < 0 || found < idx) { + idx = found + } + } + if idx < 0 { + return content, "" + } + return content[:idx], strings.TrimSpace(content[idx:]) +} + +func joinThinking(existing, added string) string { + existing = strings.TrimSpace(existing) + added = strings.TrimSpace(added) + switch { + case existing == "": + return added + case added == "": + return existing + default: + return existing + "\n" + added + } +} diff --git a/internal/providers/control_output_normalizer_test.go b/internal/providers/control_output_normalizer_test.go new file mode 100644 index 00000000..b830ecaf --- /dev/null +++ b/internal/providers/control_output_normalizer_test.go @@ -0,0 +1,313 @@ +package providers + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestOpenAIProviderChat_NormalizesKimiTextToolCall(t *testing.T) { + marker := `Dạ anh Hà ơi, em làm lại bản TTS nha!<|tool_calls_section_begin|><|tool_call_begin|>call_9e6c04c8a5df7b4aad726e9f933af3806b4<|tool_call_argument_begin|>{"text":"hello"}<|tool_call_end|><|tool_calls_section_end|>` + srv := newOpenAIJSONServer(t, fmt.Sprintf(`{ + "id":"chatcmpl-1", + "choices":[{"index":0,"message":{"role":"assistant","content":%q},"finish_reason":"stop"}] + }`, marker)) + defer srv.Close() + + p := NewOpenAIProvider("kimi-coding", "sk", srv.URL, "kimi-k2.7").WithProviderType("kimi_coding") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{{ + Type: "function", + Function: &ToolFunctionSchema{Name: "tts_synthesize"}, + }}, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if strings.Contains(resp.Content, "<|tool_calls_section_begin|>") { + t.Fatalf("raw Kimi tool marker leaked into content: %q", resp.Content) + } + if resp.Content != "Dạ anh Hà ơi, em làm lại bản TTS nha!" { + t.Fatalf("Content = %q", resp.Content) + } + if resp.FinishReason != "tool_calls" { + t.Fatalf("FinishReason = %q, want tool_calls", resp.FinishReason) + } + if len(resp.ToolCalls) != 1 { + t.Fatalf("ToolCalls = %+v, want 1", resp.ToolCalls) + } + if got := resp.ToolCalls[0].Name; got != "tts_synthesize" { + t.Fatalf("ToolCalls[0].Name = %q, want tts_synthesize", got) + } + if got := resp.ToolCalls[0].Arguments["text"]; got != "hello" { + t.Fatalf("ToolCalls[0].Arguments[text] = %v, want hello", got) + } +} + +func TestOpenAIProviderChat_InfersTextToolCallByUniqueSchemaWithMultipleTools(t *testing.T) { + marker := `prefix<|tool_calls_section_begin|><|tool_call_begin|>call_unknown<|tool_call_argument_begin|>{"text":"hello"}<|tool_call_end|><|tool_calls_section_end|>` + srv := newOpenAIJSONServer(t, fmt.Sprintf(`{ + "id":"chatcmpl-1", + "choices":[{"index":0,"message":{"role":"assistant","content":%q},"finish_reason":"stop"}] + }`, marker)) + defer srv.Close() + + p := NewOpenAIProvider("kimi-coding", "sk", srv.URL, "kimi-k2.7").WithProviderType("kimi_coding") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{ + functionToolForTest("tts", []string{"text"}), + functionToolForTest("web_search", []string{"query"}), + functionToolForTest("create_audio", []string{"prompt"}), + }, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if resp.Content != "prefix" { + t.Fatalf("Content = %q, want prefix", resp.Content) + } + if resp.FinishReason != "tool_calls" { + t.Fatalf("FinishReason = %q, want tool_calls", resp.FinishReason) + } + if len(resp.ToolCalls) != 1 { + t.Fatalf("ToolCalls = %+v, want 1", resp.ToolCalls) + } + if got := resp.ToolCalls[0].Name; got != "tts" { + t.Fatalf("ToolCalls[0].Name = %q, want tts", got) + } + if got := resp.ToolCalls[0].Arguments["text"]; got != "hello" { + t.Fatalf("ToolCalls[0].Arguments[text] = %v, want hello", got) + } +} + +func TestOpenAIProviderChat_DoesNotInferTextToolCallWhenSchemaAmbiguous(t *testing.T) { + marker := `prefix<|tool_calls_section_begin|><|tool_call_begin|>call_unknown<|tool_call_argument_begin|>{"text":"hello"}<|tool_call_end|><|tool_calls_section_end|>` + srv := newOpenAIJSONServer(t, fmt.Sprintf(`{ + "id":"chatcmpl-1", + "choices":[{"index":0,"message":{"role":"assistant","content":%q},"finish_reason":"stop"}] + }`, marker)) + defer srv.Close() + + p := NewOpenAIProvider("openai-compatible", "sk", srv.URL, "model") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{ + functionToolForTest("tts", []string{"text"}), + functionToolForTest("announce", []string{"text"}), + }, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if resp.Content != "prefix" { + t.Fatalf("Content = %q, want prefix", resp.Content) + } + if len(resp.ToolCalls) != 0 { + t.Fatalf("ToolCalls = %+v, want none for ambiguous schema match", resp.ToolCalls) + } + if resp.FinishReason == "tool_calls" { + t.Fatalf("FinishReason = %q, must not claim tool_calls when marker is ambiguous", resp.FinishReason) + } +} + +func TestOpenAIProviderChat_HandlesMixedNativeAndTextToolCalls(t *testing.T) { + marker := `also voice<|tool_calls_section_begin|><|tool_call_begin|>call_voice<|tool_call_argument_begin|>{"text":"hello"}<|tool_call_end|><|tool_calls_section_end|>` + srv := newOpenAIJSONServer(t, fmt.Sprintf(`{ + "id":"chatcmpl-1", + "choices":[{ + "index":0, + "message":{ + "role":"assistant", + "content":%q, + "tool_calls":[{ + "id":"call_datetime", + "type":"function", + "function":{"name":"datetime","arguments":"{\"timezone\":\"Asia/Ho_Chi_Minh\"}"} + }] + }, + "finish_reason":"tool_calls" + }] + }`, marker)) + defer srv.Close() + + p := NewOpenAIProvider("openai-compatible", "sk", srv.URL, "model") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{ + functionToolForTest("datetime", []string{"timezone"}), + functionToolForTest("tts", []string{"text"}), + functionToolForTest("web_search", []string{"query"}), + }, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if resp.Content != "also voice" { + t.Fatalf("Content = %q, want also voice", resp.Content) + } + if len(resp.ToolCalls) != 2 { + t.Fatalf("ToolCalls = %+v, want native datetime plus text tts", resp.ToolCalls) + } + if resp.ToolCalls[0].Name != "datetime" || resp.ToolCalls[1].Name != "tts" { + t.Fatalf("ToolCalls = %+v, want datetime then tts", resp.ToolCalls) + } +} + +func TestOpenAIProviderChat_HandlesMultipleTextToolCallsIndependently(t *testing.T) { + marker := `prefix<|tool_calls_section_begin|>` + + `<|tool_call_begin|>call_voice<|tool_call_argument_begin|>{"text":"hello"}<|tool_call_end|>` + + `<|tool_call_begin|>call_search<|tool_call_argument_begin|>{"query":"weather"}<|tool_call_end|>` + + `<|tool_calls_section_end|>` + srv := newOpenAIJSONServer(t, fmt.Sprintf(`{ + "id":"chatcmpl-1", + "choices":[{"index":0,"message":{"role":"assistant","content":%q},"finish_reason":"tool_calls"}] + }`, marker)) + defer srv.Close() + + p := NewOpenAIProvider("openai-compatible", "sk", srv.URL, "model") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{ + functionToolForTest("tts", []string{"text"}), + functionToolForTest("web_search", []string{"query"}), + functionToolForTest("create_audio", []string{"prompt"}), + }, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if resp.Content != "prefix" { + t.Fatalf("Content = %q, want prefix", resp.Content) + } + if len(resp.ToolCalls) != 2 { + t.Fatalf("ToolCalls = %+v, want 2", resp.ToolCalls) + } + if resp.ToolCalls[0].Name != "tts" || resp.ToolCalls[1].Name != "web_search" { + t.Fatalf("ToolCalls = %+v, want tts then web_search", resp.ToolCalls) + } +} + +func TestOpenAIProviderChat_DoesNotInferTextToolCallWhenArgsDoNotMatchSchema(t *testing.T) { + marker := `prefix<|tool_calls_section_begin|><|tool_call_begin|>call_unknown<|tool_call_argument_begin|>{"unknown":"hello"}<|tool_call_end|><|tool_calls_section_end|>` + srv := newOpenAIJSONServer(t, fmt.Sprintf(`{ + "id":"chatcmpl-1", + "choices":[{"index":0,"message":{"role":"assistant","content":%q},"finish_reason":"stop"}] + }`, marker)) + defer srv.Close() + + p := NewOpenAIProvider("openai-compatible", "sk", srv.URL, "model") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{ + functionToolForTest("tts", []string{"text"}), + functionToolForTest("web_search", []string{"query"}), + }, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if resp.Content != "prefix" { + t.Fatalf("Content = %q, want prefix", resp.Content) + } + if len(resp.ToolCalls) != 0 { + t.Fatalf("ToolCalls = %+v, want none when args do not match schema", resp.ToolCalls) + } +} + +func TestOpenAIProviderChat_NormalizesThinkingTags(t *testing.T) { + srv := newOpenAIJSONServer(t, `{ + "id":"chatcmpl-1", + "choices":[{"index":0,"message":{"role":"assistant","content":"hidden reasoningvisible answer"},"finish_reason":"stop"}] + }`) + defer srv.Close() + + p := NewOpenAIProvider("openai-compatible", "sk", srv.URL, "model") + resp, err := p.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "hi"}}, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + + if resp.Content != "visible answer" { + t.Fatalf("Content = %q, want visible answer", resp.Content) + } + if resp.Thinking != "hidden reasoning" { + t.Fatalf("Thinking = %q, want hidden reasoning", resp.Thinking) + } +} + +func TestOpenAIProviderChatStream_NormalizesSplitKimiTextToolCallBeforeEmittingChunks(t *testing.T) { + events := []string{ + `data: {"choices":[{"delta":{"content":"Dạ anh Hà ơi, em làm lại bản TTS nha!"}}]}` + "\n\n", + `data: {"choices":[{"delta":{"content":"<|tool_calls_section_begin|><|tool_call_begin|>call_9e6"}}]}` + "\n\n", + `data: {"choices":[{"delta":{"content":"c04c8<|tool_call_argument_begin|>{\"text\":\"hello\"}<|tool_call_end|><|tool_calls_section_end|>"}}]}` + "\n\n", + `data: {"choices":[{"finish_reason":"stop","delta":{}}]}` + "\n\n", + } + srv := newOpenAISSEServer(t, events) + defer srv.Close() + + p := NewOpenAIProvider("kimi-coding", "sk", srv.URL, "kimi-k2.7").WithProviderType("kimi_coding") + var chunks []StreamChunk + resp, err := p.ChatStream(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "say it"}}, + Tools: []ToolDefinition{{ + Type: "function", + Function: &ToolFunctionSchema{Name: "tts_synthesize"}, + }}, + }, func(chunk StreamChunk) { + chunks = append(chunks, chunk) + }) + if err != nil { + t.Fatalf("ChatStream: %v", err) + } + + for _, chunk := range chunks { + if strings.Contains(chunk.Content, "<|tool_") { + t.Fatalf("raw tool marker leaked into stream chunk: %+v", chunk) + } + } + if resp.Content != "Dạ anh Hà ơi, em làm lại bản TTS nha!" { + t.Fatalf("Content = %q", resp.Content) + } + if len(resp.ToolCalls) != 1 || resp.ToolCalls[0].Name != "tts_synthesize" { + t.Fatalf("ToolCalls = %+v, want tts_synthesize call", resp.ToolCalls) + } +} + +func newOpenAIJSONServer(t *testing.T, body string) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) + })) +} + +func functionToolForTest(name string, required []string) ToolDefinition { + properties := make(map[string]any, len(required)) + for _, key := range required { + properties[key] = map[string]any{"type": "string"} + } + return ToolDefinition{ + Type: "function", + Function: &ToolFunctionSchema{ + Name: name, + Parameters: map[string]any{ + "type": "object", + "properties": properties, + "required": required, + }, + }, + } +} diff --git a/internal/providers/openai_chat.go b/internal/providers/openai_chat.go index e25df763..084c1a8c 100644 --- a/internal/providers/openai_chat.go +++ b/internal/providers/openai_chat.go @@ -16,7 +16,7 @@ func (p *OpenAIProvider) Chat(ctx context.Context, req ChatRequest) (*ChatRespon body := p.buildRequestBody(model, req, false) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(model, req)) - chatFn := p.chatRequestFn(ctx, body) + chatFn := p.chatRequestFn(ctx, body, req.Tools) resp, err := RetryDo(ctx, p.retryConfig, chatFn) @@ -42,7 +42,7 @@ func (p *OpenAIProvider) Chat(ctx context.Context, req ChatRequest) (*ChatRespon // chatRequestFn returns a closure that performs a single non-streaming chat request. // Shared between initial attempt and post-clamp retry to avoid duplication. -func (p *OpenAIProvider) chatRequestFn(ctx context.Context, body map[string]any) func() (*ChatResponse, error) { +func (p *OpenAIProvider) chatRequestFn(ctx context.Context, body map[string]any, tools []ToolDefinition) func() (*ChatResponse, error) { return func() (*ChatResponse, error) { respBody, err := p.doRequest(ctx, body) if err != nil { @@ -55,7 +55,7 @@ func (p *OpenAIProvider) chatRequestFn(ctx context.Context, body map[string]any) return nil, fmt.Errorf("%s: decode response: %w", p.name, err) } - return p.parseResponse(&oaiResp), nil + return p.parseResponse(&oaiResp, tools), nil } } @@ -90,6 +90,8 @@ func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChun result := &ChatResponse{FinishReason: "stop"} accumulators := make(map[int]*toolCallAccumulator) + normalizer := newControlOutputNormalizer(req.Tools) + var textToolCalls []ToolCall sse := NewSSEScanner(cb) for sse.Next() { @@ -139,10 +141,9 @@ func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChun } } if delta.Content != "" { - result.Content += delta.Content - if onChunk != nil { - onChunk(StreamChunk{Content: delta.Content}) - } + normalized := normalizer.Append(delta.Content) + emitNormalizedControlChunk(result, normalized, stripThinking, onChunk) + textToolCalls = append(textToolCalls, normalized.ToolCalls...) } // Accumulate images from delta.images[]. @@ -185,6 +186,10 @@ func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChun } + normalized := normalizer.Finish() + emitNormalizedControlChunk(result, normalized, stripThinking, onChunk) + textToolCalls = append(textToolCalls, normalized.ToolCalls...) + // Check for scanner errors (timeout, connection reset, etc.) if err := sse.Err(); err != nil { return result, fmt.Errorf("%s: stream read error: %w", p.name, err) @@ -205,6 +210,7 @@ func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChun } result.ToolCalls = append(result.ToolCalls, acc.ToolCall) } + result.ToolCalls = append(result.ToolCalls, textToolCalls...) // Only override finish_reason when stream wasn't truncated. // Preserve "length" so agent loop can detect truncation and retry. diff --git a/internal/providers/openai_http.go b/internal/providers/openai_http.go index 712e5c5b..0a876590 100644 --- a/internal/providers/openai_http.go +++ b/internal/providers/openai_http.go @@ -76,8 +76,12 @@ func (p *OpenAIProvider) doRequest(ctx context.Context, body any) (io.ReadCloser return resp.Body, nil } -func (p *OpenAIProvider) parseResponse(resp *openAIResponse) *ChatResponse { +func (p *OpenAIProvider) parseResponse(resp *openAIResponse, tools ...[]ToolDefinition) *ChatResponse { result := &ChatResponse{FinishReason: "stop"} + var toolDefs []ToolDefinition + if len(tools) > 0 { + toolDefs = tools[0] + } if len(resp.Choices) > 0 { msg := resp.Choices[0].Message @@ -129,6 +133,14 @@ func (p *OpenAIProvider) parseResponse(resp *openAIResponse) *ChatResponse { Data: b64Data, }) } + + normalized := normalizeControlOutput(result.Content, result.Thinking, toolDefs) + result.Content = normalized.Content + result.Thinking = normalized.Thinking + result.ToolCalls = append(result.ToolCalls, normalized.ToolCalls...) + if len(result.ToolCalls) > 0 && result.FinishReason != "length" { + result.FinishReason = "tool_calls" + } } if resp.Usage != nil { @@ -154,6 +166,21 @@ func (p *OpenAIProvider) parseResponse(resp *openAIResponse) *ChatResponse { return result } +func emitNormalizedControlChunk(result *ChatResponse, normalized controlOutput, stripThinking bool, onChunk func(StreamChunk)) { + if normalized.Thinking != "" && !stripThinking { + result.Thinking = joinThinking(result.Thinking, normalized.Thinking) + if onChunk != nil { + onChunk(StreamChunk{Thinking: normalized.Thinking}) + } + } + if normalized.Content != "" { + result.Content += normalized.Content + if onChunk != nil { + onChunk(StreamChunk{Content: normalized.Content}) + } + } +} + // maxTokensLimitRe matches "supports at most N completion tokens" from OpenAI 400 errors. var maxTokensLimitRe = regexp.MustCompile(`supports at most (\d+) completion tokens`)