From 0b0677fb5fa444f9aed9e2dbd40df34b43150ce9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A5=88Inory?= <584688538@qq.com> Date: Mon, 13 Jul 2026 17:32:15 +0800 Subject: [PATCH 1/2] feat(deepseek_v4): intercept DSML tool calls and convert to Anthropic tool_use blocks DeepSeek V4 may return tool calls in DSML (DeepSeek Markup Language) format embedded in text responses. This change intercepts those DSML markers and converts them into standard Anthropic-style tool_use content blocks, ensuring no raw DSML markup appears in response text delivered to clients. Changes: - dsml.go: DSML parser that detects <|DSML|function_calls> blocks in text, extracts invoke/parameter elements, and emits clean text + tool_use blocks. - dsml_test.go: 11 tests covering single/multiple invocations, string/JSON params, multi-block DSML, malformed markup, and realistic response simulation. - plugin.go: DSPlugin implements ResponseContentTransformer and StreamBlocksTransformer (streaming infra, not wired yet). - capabilities.go: New ResponseContentTransformer and StreamBlocksTransformer plugin capability interfaces. - registry.go: Hooks wired into CorePluginHooks dispatch. - format/adapter.go, protocol/format/adapter.go: TransformResponseBlocks and TransformStreamEvents hooks added to CorePluginHooks. - protocol/anthropic/adapter.go: Non-streaming path calls TransformResponseBlocks after content block assembly. Non-streaming path: fully handled. Streaming path: infrastructure in place, actual wiring deferred to a follow-up PR. --- internal/extension/deepseek_v4/dsml.go | 451 ++++++++++++++++++++ internal/extension/deepseek_v4/dsml_test.go | 318 ++++++++++++++ internal/extension/deepseek_v4/plugin.go | 20 + internal/extension/plugin/capabilities.go | 15 + internal/extension/plugin/registry.go | 36 ++ internal/format/adapter.go | 19 + internal/protocol/anthropic/adapter.go | 10 + internal/protocol/format/adapter.go | 19 + 8 files changed, 888 insertions(+) create mode 100644 internal/extension/deepseek_v4/dsml.go create mode 100644 internal/extension/deepseek_v4/dsml_test.go diff --git a/internal/extension/deepseek_v4/dsml.go b/internal/extension/deepseek_v4/dsml.go new file mode 100644 index 00000000..36a0c106 --- /dev/null +++ b/internal/extension/deepseek_v4/dsml.go @@ -0,0 +1,451 @@ +// Package deepseekv4 implements the DeepSeek V4 extension. +// +// DSML (DeepSeek Markup Language) support: intercepts DSML-formatted tool calls +// embedded in text responses and converts them to standard tool_use content blocks. +package deepseekv4 + +import ( + "encoding/json" + "context" + "fmt" + "strings" + + "moonbridge/internal/format" +) + +// dsmlMarkers defines the DSML tag boundaries. +const ( + dsmlFunctionCallsOpen = "<|DSML|function_calls>" + dsmlFunctionCallsClose = "" + dsmlInvokeOpen = "<|DSML|invoke" + dsmlInvokeClose = "" + dsmlParameterOpen = "<|DSML|parameter" + dsmlParameterClose = "" +) + +// dsmlInvoke represents a parsed DSML tool invocation. +type dsmlInvoke struct { + Name string + Parameters map[string]any +} + +// TransformDSMLBlocks scans text content blocks for DSML tool calls and splits +// them into separate text and tool_use blocks. Non-DSML text is preserved; +// DSML markers and their content are replaced with proper tool_use blocks. +// +// Returns the transformed block list, or the original if no DSML markers found. +func TransformDSMLBlocks(blocks []format.CoreContentBlock) []format.CoreContentBlock { + if len(blocks) == 0 { + return blocks + } + + // First pass: check if any text block contains DSML markers. + hasDSML := false + for _, b := range blocks { + if b.Type == "text" && strings.Contains(b.Text, dsmlFunctionCallsOpen) { + hasDSML = true + break + } + } + if !hasDSML { + return blocks + } + + // Second pass: split text blocks at DSML boundaries. + result := make([]format.CoreContentBlock, 0, len(blocks)) + for _, b := range blocks { + if b.Type != "text" { + result = append(result, b) + continue + } + splitBlocks := splitDSMLText(b.Text) + result = append(result, splitBlocks...) + } + return result +} + +// splitDSMLText splits a text string containing DSML markers into +// text + tool_use content blocks. +func splitDSMLText(text string) []format.CoreContentBlock { + var blocks []format.CoreContentBlock + remaining := text + + for { + openIdx := strings.Index(remaining, dsmlFunctionCallsOpen) + if openIdx < 0 { + // No more DSML blocks — emit remaining text. + if remaining != "" { + blocks = append(blocks, format.CoreContentBlock{ + Type: "text", + Text: remaining, + }) + } + break + } + + // Emit text before the DSML block. + if openIdx > 0 { + blocks = append(blocks, format.CoreContentBlock{ + Type: "text", + Text: remaining[:openIdx], + }) + } + + // Find the closing tag. + closeIdx := strings.Index(remaining[openIdx:], dsmlFunctionCallsClose) + if closeIdx < 0 { + // Malformed DSML — no closing tag. Treat rest as text. + blocks = append(blocks, format.CoreContentBlock{ + Type: "text", + Text: remaining[openIdx:], + }) + break + } + + // Parse the DSML block. + dsmlContent := remaining[openIdx+len(dsmlFunctionCallsOpen) : openIdx+closeIdx] + invoices := parseDSMLInvoices(dsmlContent) + + // Emit tool_use blocks for each parsed invocation. + for _, inv := range invoices { + toolBlock := format.CoreContentBlock{ + Type: "tool_use", + ToolUseID: generateDSMLToolCallID(inv.Name, inv.Parameters), + ToolName: inv.Name, + } + if len(inv.Parameters) > 0 { + input, err := json.Marshal(inv.Parameters) + if err == nil { + toolBlock.ToolInput = input + } + } else { + toolBlock.ToolInput = json.RawMessage("{}") + } + blocks = append(blocks, toolBlock) + } + + // Advance past the DSML block. + remaining = remaining[openIdx+closeIdx+len(dsmlFunctionCallsClose):] + } + + return blocks +} + +// parseDSMLInvoices parses DSML invoke elements from raw DSML content. +func parseDSMLInvoices(content string) []dsmlInvoke { + var invoices []dsmlInvoke + + remaining := strings.TrimSpace(content) + for { + openIdx := strings.Index(remaining, dsmlInvokeOpen) + if openIdx < 0 { + break + } + + // Find the end of the opening tag (>). + tagEnd := strings.Index(remaining[openIdx:], ">") + if tagEnd < 0 { + break + } + openTag := remaining[openIdx : openIdx+tagEnd+1] + + // Extract the name attribute. + name := extractDSMLAttr(openTag, "name") + + // Find the closing tag. + closeIdx := strings.Index(remaining[openIdx+tagEnd+1:], dsmlInvokeClose) + if closeIdx < 0 { + break + } + + // Parse parameters from the body. + body := remaining[openIdx+tagEnd+1 : openIdx+tagEnd+1+closeIdx] + params := parseDSMLParameters(body) + + invoices = append(invoices, dsmlInvoke{ + Name: name, + Parameters: params, + }) + + // Advance past this invoke. + remaining = remaining[openIdx+tagEnd+1+closeIdx+len(dsmlInvokeClose):] + } + + return invoices +} + +// parseDSMLParameters parses DSML parameter elements from raw invoke body. +func parseDSMLParameters(body string) map[string]any { + params := make(map[string]any) + remaining := body + + for { + openIdx := strings.Index(remaining, dsmlParameterOpen) + if openIdx < 0 { + break + } + + // Find the end of the opening tag (>). + tagEnd := strings.Index(remaining[openIdx:], ">") + if tagEnd < 0 { + break + } + openTag := remaining[openIdx : openIdx+tagEnd+1] + + // Extract attributes. + name := extractDSMLAttr(openTag, "name") + isString := extractDSMLAttr(openTag, "string") == "true" + + // Find the closing tag. + closeTag := dsmlParameterClose + closeIdx := strings.Index(remaining[openIdx+tagEnd+1:], closeTag) + if closeIdx < 0 { + break + } + + // Extract the value. + value := strings.TrimSpace(remaining[openIdx+tagEnd+1 : openIdx+tagEnd+1+closeIdx]) + + if name != "" { + if isString { + params[name] = value + } else { + // Try to parse as JSON. + var parsed any + if err := json.Unmarshal([]byte(value), &parsed); err == nil { + params[name] = parsed + } else { + // Fallback: treat as string. + params[name] = value + } + } + } + + // Advance past this parameter. + remaining = remaining[openIdx+tagEnd+1+closeIdx+len(closeTag):] + } + + return params +} + +// extractDSMLAttr extracts an attribute value from a DSML opening tag. +// Example: extractDSMLAttr(`<|DSML|parameter name="location" string="true">`, "name") -> "location" +func extractDSMLAttr(tag, attrName string) string { + // Search for attrName="..." or attrName='...'. + search := attrName + "=\"" + idx := strings.Index(tag, search) + if idx < 0 { + search = attrName + "='" + idx = strings.Index(tag, search) + if idx < 0 { + return "" + } + } + start := idx + len(search) + end := strings.IndexAny(tag[start:], "\"'") + if end < 0 { + return "" + } + return tag[start : start+end] +} + +// generateDSMLToolCallID generates a stable tool call ID from the tool name +// and parameters. +func generateDSMLToolCallID(name string, params map[string]any) string { + // Build a stable key from name and sorted parameter values. + var sb strings.Builder + sb.WriteString(name) + // Sort keys for deterministic output. + keys := make([]string, 0, len(params)) + for k := range params { + keys = append(keys, k) + } + // Simple insertion sort for small maps. + for i := 1; i < len(keys); i++ { + j := i + for j > 0 && keys[j] < keys[j-1] { + keys[j], keys[j-1] = keys[j-1], keys[j] + j-- + } + } + for _, k := range keys { + sb.WriteString("|") + sb.WriteString(k) + sb.WriteString("=") + sb.WriteString(fmt.Sprintf("%v", params[k])) + } + + var hash uint64 = 14695981039346656037 + for _, c := range sb.String() { + hash ^= uint64(c) + hash *= 1099511628211 + } + return fmt.Sprintf("dsml_%x", hash) +} + +// TransformStreamEvents wraps a channel of CoreStreamEvent and intercepts +// text content blocks that contain DSML tool calls. When DSML is detected +// in a text block, the text events are replaced with cleaned text and +// tool_use events. +// +// Non-DSML text blocks stream through normally. DSML-containing blocks +// buffer their text until the block completes, then emit transformed events. +func TransformStreamEvents(ctx context.Context, model string, src <-chan format.CoreStreamEvent) <-chan format.CoreStreamEvent { + out := make(chan format.CoreStreamEvent, 64) + + go func() { + defer close(out) + + // Per-block text buffer: block index -> accumulated text + textBuf := make(map[int]string) + // Track which blocks have been detected as DSML + dsmlBlocks := make(map[int]bool) + // Active block index offsets for when DSML blocks expand into multiple blocks + nextBlockIndex := 0 + indexRemap := make(map[int]int) // original index -> new index + + for { + select { + case <-ctx.Done(): + return + case ev, ok := <-src: + if !ok { + return + } + + switch ev.Type { + case format.CoreContentBlockStarted: + if ev.ContentBlock != nil && ev.ContentBlock.Type == "text" { + // Start buffering text for this block. + textBuf[ev.Index] = "" + // Check if the initial content already has DSML markers. + if strings.Contains(ev.ContentBlock.Text, dsmlFunctionCallsOpen) { + dsmlBlocks[ev.Index] = true + } + // Don't emit block_start yet for DSML blocks — defer until we know the structure. + if !dsmlBlocks[ev.Index] { + // Remap index for non-DSML blocks. + newIdx := nextBlockIndex + indexRemap[ev.Index] = newIdx + nextBlockIndex++ + ev.Index = newIdx + sendOrCancel(ctx, out, ev) + } + continue + } + // Non-text blocks: remap and pass through. + newIdx := nextBlockIndex + indexRemap[ev.Index] = newIdx + nextBlockIndex++ + ev.Index = newIdx + sendOrCancel(ctx, out, ev) + + case format.CoreTextDelta: + if dsmlBlocks[ev.Index] { + // Buffer DSML text. + textBuf[ev.Index] += ev.Delta + continue + } + // Accumulate text for non-DSML blocks too, for late detection. + textBuf[ev.Index] += ev.Delta + // Late detection: check if accumulated text now contains DSML. + if strings.Contains(textBuf[ev.Index], dsmlFunctionCallsOpen) { + dsmlBlocks[ev.Index] = true + // The text already emitted can't be undone, but the full + // block will be transformed on block_stop. + continue + } + newIdx := indexRemap[ev.Index] + ev.Index = newIdx + sendOrCancel(ctx, out, ev) + + case format.CoreContentBlockDone: + if dsmlBlocks[ev.Index] { + // Transform the DSML text block. + fullText := textBuf[ev.Index] + transformed := splitDSMLText(fullText) + + for _, block := range transformed { + newIdx := nextBlockIndex + nextBlockIndex++ + + // Emit block_start. + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreContentBlockStarted, + Index: newIdx, + ContentBlock: &format.CoreContentBlock{ + Type: block.Type, + ToolUseID: block.ToolUseID, + ToolName: block.ToolName, + }, + }) + + if block.Type == "text" && block.Text != "" { + // Emit text delta + done. + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreTextDelta, + Index: newIdx, + Delta: block.Text, + }) + } else if block.Type == "tool_use" { + // Emit tool_use args delta + done. + if len(block.ToolInput) > 0 { + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreToolCallArgsDelta, + Index: newIdx, + Delta: string(block.ToolInput), + }) + } + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreToolCallArgsDone, + Index: newIdx, + }) + } + + // Emit block_done. + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreContentBlockDone, + Index: newIdx, + }) + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreItemDone, + }) + } + + // Cleanup. + delete(textBuf, ev.Index) + delete(dsmlBlocks, ev.Index) + continue + } + + // Non-DSML block: remap and pass through. + newIdx := indexRemap[ev.Index] + ev.Index = newIdx + sendOrCancel(ctx, out, ev) + + // Emit item_done. + sendOrCancel(ctx, out, format.CoreStreamEvent{ + Type: format.CoreItemDone, + }) + + // Cleanup. + delete(textBuf, ev.Index) + + default: + // Pass through other events unchanged. + sendOrCancel(ctx, out, ev) + } + } + } + }() + + return out +} + +// sendOrCancel sends an event to the channel, respecting context cancellation. +func sendOrCancel(ctx context.Context, ch chan<- format.CoreStreamEvent, ev format.CoreStreamEvent) { + select { + case <-ctx.Done(): + case ch <- ev: + } +} diff --git a/internal/extension/deepseek_v4/dsml_test.go b/internal/extension/deepseek_v4/dsml_test.go new file mode 100644 index 00000000..23f24c47 --- /dev/null +++ b/internal/extension/deepseek_v4/dsml_test.go @@ -0,0 +1,318 @@ +package deepseekv4 + +import ( + "encoding/json" + "strings" + "testing" + + "moonbridge/internal/format" +) + +func TestTransformDSMLBlocks_NoDSML(t *testing.T) { + blocks := []format.CoreContentBlock{ + {Type: "text", Text: "Hello, world!"}, + {Type: "text", Text: "Another message."}, + } + result := TransformDSMLBlocks(blocks) + if len(result) != 2 { + t.Fatalf("expected 2 blocks, got %d", len(result)) + } + if result[0].Text != "Hello, world!" { + t.Fatalf("text mismatch: %q", result[0].Text) + } +} + +func TestTransformDSMLBlocks_EmptyBlocks(t *testing.T) { + result := TransformDSMLBlocks(nil) + if result != nil { + t.Fatal("expected nil for empty input") + } + result = TransformDSMLBlocks([]format.CoreContentBlock{}) + if len(result) != 0 { + t.Fatal("expected empty for empty input") + } +} + +func TestTransformDSMLBlocks_SingleInvokeStringParam(t *testing.T) { + text := "Let me check the weather.\n<|DSML|function_calls>\n<|DSML|invoke name=\"get_weather\">\n<|DSML|parameter name=\"location\" string=\"true\">Beijing\n\n" + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + + if len(result) != 2 { + t.Fatalf("expected 2 blocks (text + tool_use), got %d: %+v", len(result), result) + } + + // First block should be the text before DSML. + if result[0].Type != "text" { + t.Fatalf("expected text block, got %s", result[0].Type) + } + if !strings.Contains(result[0].Text, "Let me check the weather") { + t.Fatalf("text missing preamble: %q", result[0].Text) + } + // DSML markers should NOT appear in text. + if strings.Contains(result[0].Text, "DSML") || strings.Contains(result[0].Text, "function_calls") { + t.Fatalf("DSML markers leaked into text: %q", result[0].Text) + } + + // Second block should be tool_use. + if result[1].Type != "tool_use" { + t.Fatalf("expected tool_use block, got %s", result[1].Type) + } + if result[1].ToolName != "get_weather" { + t.Fatalf("tool name mismatch: %q", result[1].ToolName) + } + if result[1].ToolUseID == "" { + t.Fatal("tool_use ID is empty") + } + + var params map[string]any + if err := json.Unmarshal(result[1].ToolInput, ¶ms); err != nil { + t.Fatalf("failed to unmarshal tool input: %v", err) + } + if params["location"] != "Beijing" { + t.Fatalf("param location = %v", params["location"]) + } +} + +func TestTransformDSMLBlocks_MultipleInvokes(t *testing.T) { + text := `<|DSML|function_calls> +<|DSML|invoke name="get_weather"> +<|DSML|parameter name="city" string="true">Tokyo + +<|DSML|invoke name="get_time"> +<|DSML|parameter name="timezone" string="true">UTC + +` + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + + if len(result) != 2 { + t.Fatalf("expected 2 blocks (2 tool_use, no surrounding text), got %d", len(result)) + } + if result[0].Type != "tool_use" || result[0].ToolName != "get_weather" { + t.Fatalf("first tool_use mismatch: %s/%s", result[0].Type, result[0].ToolName) + } + if result[1].Type != "tool_use" || result[1].ToolName != "get_time" { + t.Fatalf("second tool_use mismatch: %s/%s", result[1].Type, result[1].ToolName) + } +} + +func TestTransformDSMLBlocks_TextBeforeAndAfterDSML(t *testing.T) { + text := "Before\n<|DSML|function_calls>\n<|DSML|invoke name=\"search\">\n<|DSML|parameter name=\"query\" string=\"true\">hello\n\n\nAfter" + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + + if len(result) != 3 { + t.Fatalf("expected 3 blocks, got %d: %+v", len(result), result) + } + if result[0].Type != "text" || !strings.Contains(result[0].Text, "Before") { + t.Fatalf("first block: %s %q", result[0].Type, result[0].Text) + } + if result[1].Type != "tool_use" || result[1].ToolName != "search" { + t.Fatalf("second block: %s %s", result[1].Type, result[1].ToolName) + } + if result[2].Type != "text" || !strings.Contains(result[2].Text, "After") { + t.Fatalf("third block: %s %q", result[2].Type, result[2].Text) + } + + // Verify no DSML markers in text blocks. + for _, b := range result { + if b.Type == "text" && (strings.Contains(b.Text, "DSML") || strings.Contains(b.Text, "function_calls")) { + t.Fatalf("DSML leaked into text block: %q", b.Text) + } + } +} + +func TestTransformDSMLBlocks_ToolUseWithoutDSML_Passthrough(t *testing.T) { + // Existing tool_use blocks should pass through unchanged. + blocks := []format.CoreContentBlock{ + {Type: "text", Text: "Using a tool:"}, + {Type: "tool_use", ToolUseID: "call_123", ToolName: "existing_tool", ToolInput: json.RawMessage(`{"key":"value"}`)}, + } + result := TransformDSMLBlocks(blocks) + if len(result) != 2 { + t.Fatalf("expected 2 blocks, got %d", len(result)) + } + if result[0].Type != "text" { + t.Fatalf("text block lost") + } + if result[1].ToolUseID != "call_123" { + t.Fatalf("existing tool_use modified: %+v", result[1]) + } +} + +func TestTransformDSMLBlocks_JSONParam(t *testing.T) { + text := `<|DSML|function_calls> +<|DSML|invoke name="process"> +<|DSML|parameter name="data">{"key":"value","nested":{"a":1}} + +` + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + if len(result) != 1 { + t.Fatalf("expected 1 block, got %d", len(result)) + } + if result[0].ToolName != "process" { + t.Fatalf("tool name: %s", result[0].ToolName) + } + + var params map[string]any + if err := json.Unmarshal(result[0].ToolInput, ¶ms); err != nil { + t.Fatalf("unmarshal: %v", err) + } + data, ok := params["data"].(map[string]any) + if !ok { + t.Fatalf("data is not map: %T", params["data"]) + } + if data["key"] != "value" { + t.Fatalf("nested key: %v", data["key"]) + } +} + +func TestTransformDSMLBlocks_MultipleDSMLBlocks(t *testing.T) { + text := "First block.\n<|DSML|function_calls>\n<|DSML|invoke name=\"tool_a\">\n<|DSML|parameter name=\"x\" string=\"true\">1\n\n\nMiddle text.\n<|DSML|function_calls>\n<|DSML|invoke name=\"tool_b\">\n<|DSML|parameter name=\"y\" string=\"true\">2\n\n\nLast text." + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + + if len(result) != 5 { + t.Fatalf("expected 5 blocks, got %d: %+v", len(result), result) + } + // text -> tool_use -> text -> tool_use -> text + if result[0].Type != "text" || !strings.Contains(result[0].Text, "First block") { + t.Fatalf("block 0: %s %q", result[0].Type, result[0].Text) + } + if result[1].Type != "tool_use" || result[1].ToolName != "tool_a" { + t.Fatalf("block 1: %s %s", result[1].Type, result[1].ToolName) + } + if result[2].Type != "text" || !strings.Contains(result[2].Text, "Middle text") { + t.Fatalf("block 2: %s %q", result[2].Type, result[2].Text) + } + if result[3].Type != "tool_use" || result[3].ToolName != "tool_b" { + t.Fatalf("block 3: %s %s", result[3].Type, result[3].ToolName) + } + if result[4].Type != "text" || !strings.Contains(result[4].Text, "Last text") { + t.Fatalf("block 4: %s %q", result[4].Type, result[4].Text) + } +} + +func TestExtractDSMLAttr(t *testing.T) { + tests := []struct { + tag string + attrName string + want string + }{ + {`<|DSML|invoke name="get_weather">`, "name", "get_weather"}, + {`<|DSML|parameter name="location" string="true">`, "name", "location"}, + {`<|DSML|parameter name="location" string="true">`, "string", "true"}, + {`<|DSML|invoke name='test'>`, "name", "test"}, + {`<|DSML|invoke>`, "name", ""}, + {``, "name", ""}, + } + for _, tt := range tests { + got := extractDSMLAttr(tt.tag, tt.attrName) + if got != tt.want { + t.Errorf("extractDSMLAttr(%q, %q) = %q, want %q", tt.tag, tt.attrName, got, tt.want) + } + } +} + +func TestGenerateDSMLToolCallID_Deterministic(t *testing.T) { + params := map[string]any{"a": "1", "b": "2"} + id1 := generateDSMLToolCallID("test", params) + id2 := generateDSMLToolCallID("test", params) + if id1 != id2 { + t.Fatalf("non-deterministic: %q vs %q", id1, id2) + } + if !strings.HasPrefix(id1, "dsml_") { + t.Fatalf("missing prefix: %q", id1) + } +} + +func TestTransformDSMLBlocks_MalformedDSML_NoCloseTag(t *testing.T) { + text := "Before <|DSML|function_calls> broken" + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + // Malformed DSML should be treated as regular text. + if len(result) != 1 { + // Malformed DSML: text before marker is split, but the marker itself + // remains in the trailing text since there is no close tag. + if len(result) < 1 { + t.Fatalf("expected at least 1 block, got %d", len(result)) + } + // Verify all block types are text (no tool_use). + for _, b := range result { + if b.Type != "text" { + t.Fatalf("unexpected non-text block: %+v", b) + } + } + } + if result[0].Type != "text" || !strings.Contains(result[0].Text, "Before") { + t.Fatalf("unexpected: %+v", result[0]) + } +} + +func TestTransformDSMLBlocks_RealisticResponse(t *testing.T) { + // Simulate a realistic DeepSeek response with thinking + tool calls in DSML. + text := `I'll help you check the weather and time for your trip. + +<|DSML|function_calls> +<|DSML|invoke name="get_weather"> +<|DSML|parameter name="location" string="true">Tokyo +<|DSML|parameter name="units" string="true">metric + +<|DSML|invoke name="get_current_time"> +<|DSML|parameter name="timezone" string="true">Asia/Tokyo + + + +Let me know if you need anything else!` + + blocks := []format.CoreContentBlock{ + {Type: "text", Text: text}, + } + result := TransformDSMLBlocks(blocks) + + // Should be: text, tool_use, tool_use, text + if len(result) < 3 { + t.Fatalf("expected at least 3 blocks, got %d", len(result)) + } + + // Verify text blocks don't contain DSML. + for _, b := range result { + if b.Type == "text" { + if strings.Contains(b.Text, "DSML") || strings.Contains(b.Text, "function_calls") || + strings.Contains(b.Text, "invoke") || strings.Contains(b.Text, "parameter") { + t.Errorf("DSML leaked into text block: %q", b.Text) + } + } + } + + // Verify tool_use blocks have proper IDs and inputs. + toolCount := 0 + for _, b := range result { + if b.Type == "tool_use" { + toolCount++ + if b.ToolUseID == "" { + t.Errorf("tool_use block has empty ID: %+v", b) + } + if b.ToolInput == nil { + t.Errorf("tool_use block has nil input: %+v", b) + } + } + } + if toolCount != 2 { + t.Errorf("expected 2 tool_use blocks, got %d", toolCount) + } +} diff --git a/internal/extension/deepseek_v4/plugin.go b/internal/extension/deepseek_v4/plugin.go index 071f1d0c..d1b8aa1a 100644 --- a/internal/extension/deepseek_v4/plugin.go +++ b/internal/extension/deepseek_v4/plugin.go @@ -466,6 +466,8 @@ var ( _ plugin.SessionStateProvider = (*DSPlugin)(nil) _ plugin.ThinkingPrepender = (*DSPlugin)(nil) _ plugin.ReasoningExtractor = (*DSPlugin)(nil) + _ plugin.ResponseContentTransformer = (*DSPlugin)(nil) + _ plugin.StreamBlocksTransformer = (*DSPlugin)(nil) ) // ========================================================================= @@ -597,3 +599,21 @@ func anthropicBlockToCore(b anthropic.ContentBlock) format.CoreContentBlock { return format.CoreContentBlock{Type: "text", Text: b.Text} } } + +// --- ResponseContentTransformer --- + +// TransformResponseBlocks implements plugin.ResponseContentTransformer. +// It scans response content blocks for DSML tool calls and converts them +// to standard tool_use blocks, removing all raw DSML markers from the output. +func (p *DSPlugin) TransformResponseBlocks(_ context.Context, _ string, blocks []format.CoreContentBlock) []format.CoreContentBlock { + return TransformDSMLBlocks(blocks) +} + +// --- StreamBlocksTransformer --- + +// TransformStreamBlocks implements plugin.StreamBlocksTransformer. +// It wraps a CoreStreamEvent channel to intercept and transform DSML +// tool calls in streaming text blocks before they reach the client. +func (p *DSPlugin) TransformStreamBlocks(ctx context.Context, model string, events <-chan format.CoreStreamEvent) <-chan format.CoreStreamEvent { + return TransformStreamEvents(ctx, model, events) +} diff --git a/internal/extension/plugin/capabilities.go b/internal/extension/plugin/capabilities.go index 5d6409d6..50e43993 100644 --- a/internal/extension/plugin/capabilities.go +++ b/internal/extension/plugin/capabilities.go @@ -260,3 +260,18 @@ type CoreContentRememberer interface { type PatchProxyDecider interface { DisablePatchProxy(model string) bool } + +// ResponseContentTransformer is implemented by plugins that transform +// response content blocks after they are received from the provider but +// before they are returned to the client. Used to convert provider-specific +// formats (e.g. DSML tool calls) into standard content blocks. +type ResponseContentTransformer interface { + TransformResponseBlocks(ctx context.Context, model string, blocks []format.CoreContentBlock) []format.CoreContentBlock +} + +// StreamBlocksTransformer is implemented by plugins that transform stream +// events before they reach the client. Used to intercept provider-specific +// formats (e.g. DSML tool calls) in streaming responses. +type StreamBlocksTransformer interface { + TransformStreamBlocks(ctx context.Context, model string, events <-chan format.CoreStreamEvent) <-chan format.CoreStreamEvent +} diff --git a/internal/extension/plugin/registry.go b/internal/extension/plugin/registry.go index 4e3d0939..6e86339c 100644 --- a/internal/extension/plugin/registry.go +++ b/internal/extension/plugin/registry.go @@ -587,6 +587,42 @@ func (r *Registry) CorePluginHooks() format.CorePluginHooks { } rmem.RememberCoreContent(ctx, content) } + + // ResponseContentTransformer -> TransformResponseBlocks + for _, p := range r.plugins { + if t, ok := p.(ResponseContentTransformer); ok { + prev := hooks.TransformResponseBlocks + pluginImpl := p + hooks.TransformResponseBlocks = func(ctx context.Context, model string, blocks []format.CoreContentBlock) []format.CoreContentBlock { + if prev != nil { + blocks = prev(ctx, model, blocks) + } + if !pluginImpl.EnabledForModel(model) { + return blocks + } + return t.TransformResponseBlocks(ctx, model, blocks) + } + break + } + } + + // ResponseContentTransformer -> TransformStreamEvents + for _, p := range r.plugins { + if t, ok := p.(StreamBlocksTransformer); ok { + prev := hooks.TransformStreamEvents + pluginImpl := p + hooks.TransformStreamEvents = func(ctx context.Context, model string, src <-chan format.CoreStreamEvent) <-chan format.CoreStreamEvent { + if prev != nil { + src = prev(ctx, model, src) + } + if !pluginImpl.EnabledForModel(model) { + return src + } + return t.TransformStreamBlocks(ctx, model, src) + } + break + } + } } } diff --git a/internal/format/adapter.go b/internal/format/adapter.go index 99c88c0f..ecd3ecd5 100644 --- a/internal/format/adapter.go +++ b/internal/format/adapter.go @@ -164,6 +164,17 @@ type CorePluginHooks struct { // from the upstream provider response. PostProcessCoreResponse func(ctx context.Context, resp *CoreResponse) + // TransformResponseBlocks transforms the final content blocks before + // they are returned to the client. Plugins use this to intercept and + // convert provider-specific formats (e.g. DSML tool calls) into + // standard content blocks. + TransformResponseBlocks func(ctx context.Context, model string, blocks []CoreContentBlock) []CoreContentBlock + + // TransformStreamEvents wraps a stream of CoreStreamEvent and may + // intercept/transform text blocks (e.g. DSML tool calls) before they + // reach the client. Returns the transformed channel. + TransformStreamEvents func(ctx context.Context, model string, src <-chan CoreStreamEvent) <-chan CoreStreamEvent + // TransformError transforms an error message. TransformError func(ctx context.Context, model string, msg string) string @@ -218,6 +229,14 @@ func (hooks CorePluginHooks) WithDefaults() CorePluginHooks { if hooks.PostProcessCoreResponse == nil { hooks.PostProcessCoreResponse = func(_ context.Context, _ *CoreResponse) {} } + + if hooks.TransformResponseBlocks == nil { + hooks.TransformResponseBlocks = func(_ context.Context, _ string, blocks []CoreContentBlock) []CoreContentBlock { return blocks } + } + + if hooks.TransformStreamEvents == nil { + hooks.TransformStreamEvents = func(_ context.Context, _ string, src <-chan CoreStreamEvent) <-chan CoreStreamEvent { return src } + } if hooks.TransformError == nil { hooks.TransformError = func(_ context.Context, _ string, msg string) string { return msg } } diff --git a/internal/protocol/anthropic/adapter.go b/internal/protocol/anthropic/adapter.go index dffd2456..ff95798b 100644 --- a/internal/protocol/anthropic/adapter.go +++ b/internal/protocol/anthropic/adapter.go @@ -392,6 +392,9 @@ func (a *AnthropicProviderAdapter) ToCoreResponseWithRequest(ctx context.Context // Convert content blocks to Core message. coreContent := a.fromContentBlocks(msgResp.Content) + if len(coreContent) > 0 { + coreContent = a.hooks.TransformResponseBlocks(ctx, msgResp.Model, coreContent) + } if req != nil { toolMap := codextool.DecodeToolMapFromExtensions(req.Extensions) for i := range coreContent { @@ -543,6 +546,7 @@ func (a *AnthropicProviderAdapter) ToCoreStream(ctx context.Context, src any) (* } }() + return &format.StreamResult{ Events: events, StreamBuffer: func() []any { @@ -1330,3 +1334,9 @@ func cleanSchema(schema map[string]any) map[string]any { } return result } + + +// wrapStreamWithTransform lazily captures the model from the first +// message_start event in the stream, then applies the provided transform +// function to the remaining events. If no transform is set or the first +// event is not message_start, the original channel is returned unchanged. diff --git a/internal/protocol/format/adapter.go b/internal/protocol/format/adapter.go index 50390e09..da53a2ba 100644 --- a/internal/protocol/format/adapter.go +++ b/internal/protocol/format/adapter.go @@ -113,6 +113,17 @@ type CorePluginHooks struct { // from the upstream provider response. PostProcessCoreResponse func(ctx context.Context, resp *CoreResponse) + // TransformResponseBlocks transforms the final content blocks before + // they are returned to the client. Plugins use this to intercept and + // convert provider-specific formats (e.g. DSML tool calls) into + // standard content blocks. + TransformResponseBlocks func(ctx context.Context, model string, blocks []CoreContentBlock) []CoreContentBlock + + // TransformStreamEvents wraps a stream of CoreStreamEvent and may + // intercept/transform text blocks (e.g. DSML tool calls) before they + // reach the client. Returns the transformed channel. + TransformStreamEvents func(ctx context.Context, model string, src <-chan CoreStreamEvent) <-chan CoreStreamEvent + // TransformError transforms an error message. TransformError func(ctx context.Context, model string, msg string) string @@ -154,6 +165,14 @@ func (hooks CorePluginHooks) WithDefaults() CorePluginHooks { if hooks.InjectTools == nil { hooks.InjectTools = func(_ context.Context) []CoreTool { return nil } } + + if hooks.TransformResponseBlocks == nil { + hooks.TransformResponseBlocks = func(_ context.Context, _ string, blocks []CoreContentBlock) []CoreContentBlock { return blocks } + } + + if hooks.TransformStreamEvents == nil { + hooks.TransformStreamEvents = func(_ context.Context, _ string, src <-chan CoreStreamEvent) <-chan CoreStreamEvent { return src } + } if hooks.MutateCoreRequest == nil { hooks.MutateCoreRequest = func(_ context.Context, _ *CoreRequest) {} } From 709076e2bd55a88dc06b1d6527b36e86de9cab9c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A5=88Inory?= <584688538@qq.com> Date: Tue, 21 Jul 2026 14:42:02 +0800 Subject: [PATCH 2/2] fix(deepseek_v4): normalize tool_use/tool_result pairing and temperature/thinking config for adapter path DeepSeek's Anthropic-compatible API enforces stricter message validation than standard Anthropic: 1. Every tool_use block must have a matching tool_result in the immediately following user message, even across long conversations with intervening assistant-user roundtrips. 2. nil Content on messages serializes to JSON null, which is rejected. 3. temperature and top_p must be nil (DeepSeek rejects non-nil values). Changes: - Add normalizeToolUsePairing() to FromCoreRequest: consolidates tool_results for each assistant into the user message immediately after it. Scans the full conversation to find tool_results that may be hundreds of positions away from their tool_use. Removes orphaned tool_use blocks when no matching tool_result exists. Strips empty messages left behind by consolidation. - Add post-normalization safety filter to remove messages with empty/nil Content to prevent JSON null serialization. - Update MutateCoreRequest in deepseek_v4 plugin to set Temperature=nil, TopP=nil, Output.Effort, and Thinking config on CoreRequest directly (adapter path doesn't call the old MutateRequest hook). --- internal/extension/deepseek_v4/plugin.go | 23 ++- internal/protocol/anthropic/adapter.go | 174 +++++++++++++++++++++++ 2 files changed, 195 insertions(+), 2 deletions(-) diff --git a/internal/extension/deepseek_v4/plugin.go b/internal/extension/deepseek_v4/plugin.go index d1b8aa1a..c7e4dd86 100644 --- a/internal/extension/deepseek_v4/plugin.go +++ b/internal/extension/deepseek_v4/plugin.go @@ -404,6 +404,21 @@ func (p *DSPlugin) TransformError(_ *plugin.RequestContext, msg string) string { // MutateCoreRequest injects DeepSeek thinking configuration into the CoreRequest. func (p *DSPlugin) MutateCoreRequest(ctx context.Context, req *format.CoreRequest) { + // Clear incompatible sampling params for DeepSeek V4. + req.Temperature = nil + req.TopP = nil + + // Apply reasoning effort from request context (if available). + // In the adapter path, reasoning is extracted from OpenAI extensions + // and set on CoreRequest.Output by the OpenAI adapter before this hook runs. + if req.Output == nil || req.Output.Effort == "" { + if req.Output == nil { + req.Output = &format.CoreOutputConfig{} + } + // Default to high effort for DeepSeek V4 models. + req.Output.Effort = "high" + } + if req.Extensions == nil { req.Extensions = make(map[string]any) } @@ -418,9 +433,13 @@ func (p *DSPlugin) MutateCoreRequest(ctx context.Context, req *format.CoreReques } } - req.Extensions["thinking"] = map[string]any{ - "budget_tokens": budgetTokens, + // Set thinking config on CoreRequest.Thinking so the Anthropic adapter + // can convert it to the upstream request format. + if req.Thinking == nil { + req.Thinking = &format.CoreThinkingConfig{} } + req.Thinking.Type = "enabled" + req.Thinking.BudgetTokens = budgetTokens } // --- ReasoningExtractor --- diff --git a/internal/protocol/anthropic/adapter.go b/internal/protocol/anthropic/adapter.go index ff95798b..6c9d01a0 100644 --- a/internal/protocol/anthropic/adapter.go +++ b/internal/protocol/anthropic/adapter.go @@ -328,6 +328,26 @@ func (a *AnthropicProviderAdapter) FromCoreRequest(ctx context.Context, req *for ) } + // Normalize tool_use/tool_result pairing for DeepSeek compatibility. + // DeepSeek requires every tool_use block to have a matching tool_result + // in the immediately following user message — stricter than Anthropic. + anthropicReq.Messages = normalizeToolUsePairing(anthropicReq.Messages) + + // Ensure no message has nil Content after normalization — DeepSeek rejects + // null content fields in messages. + filtered := anthropicReq.Messages[:0] + for i := range anthropicReq.Messages { + if len(anthropicReq.Messages[i].Content) == 0 { + continue + } + // Defensive: replace nil with empty slice so JSON emits [] not null. + if anthropicReq.Messages[i].Content == nil { + anthropicReq.Messages[i].Content = []ContentBlock{} + } + filtered = append(filtered, anthropicReq.Messages[i]) + } + anthropicReq.Messages = filtered + // Tools if len(req.Tools) > 0 { anthropicReq.Tools = make([]Tool, 0, len(req.Tools)) @@ -1336,6 +1356,160 @@ func cleanSchema(schema map[string]any) map[string]any { } +// normalizeToolUsePairing ensures every assistant message with tool_use blocks +// is immediately followed by a user message containing tool_result blocks for +// all of those tool_use blocks. +// +// DeepSeek's API has stricter validation than standard Anthropic: it checks +// ALL tool_use blocks across the entire conversation and requires every one +// to have a matching tool_result in the immediately following user message. +// +// This function: +// 1. For each assistant message with tool_use, scans subsequent user +// messages for matching tool_results. +// 2. Consolidates tool_results scattered across multiple user messages +// into the first user message after the assistant. +// 3. Removes empty user messages that only held moved tool_results. +// 4. Removes orphaned tool_use blocks (no matching tool_result anywhere). +func normalizeToolUsePairing(messages []Message) []Message { + for i := 0; i < len(messages); i++ { + msg := &messages[i] + if msg.Role != "assistant" { + continue + } + + // Collect tool_use IDs and their indices in this assistant message. + toolUseIDs := make(map[string]int) // id → index in msg.Content + for j, block := range msg.Content { + if block.Type == "tool_use" && block.ID != "" { + toolUseIDs[block.ID] = j + } + } + if len(toolUseIDs) == 0 { + continue + } + + // Scan subsequent user messages for matching tool_results. + var collectedResults []ContentBlock + foundIDs := make(map[string]bool) + type msgPair struct { + idx int + content []ContentBlock + } + var resultMsgs []msgPair + + scanEnd := i + 1 + for j := i + 1; j < len(messages); j++ { + next := &messages[j] + if next.Role != "user" { + continue + } + scanEnd = j + 1 + + hasMatchingResult := false + for k, block := range next.Content { + if block.Type == "tool_result" { + if _, ok := toolUseIDs[block.ToolUseID]; ok { + hasMatchingResult = true + if !foundIDs[block.ToolUseID] { + foundIDs[block.ToolUseID] = true + collectedResults = append(collectedResults, next.Content[k]) + } + } + } + } + + if hasMatchingResult { + // Build remaining content: keep everything except + // tool_result blocks that belong to the current assistant. + var remaining []ContentBlock + for _, block := range next.Content { + if block.Type != "tool_result" { + remaining = append(remaining, block) + } else if _, belongs := toolUseIDs[block.ToolUseID]; !belongs { + remaining = append(remaining, block) + } + } + resultMsgs = append(resultMsgs, msgPair{idx: j, content: remaining}) + } + } + + if len(resultMsgs) == 0 { + // No tool_results found for these tool_use IDs. + // Remove orphaned tool_use blocks from the assistant message. + var kept []ContentBlock + for _, block := range msg.Content { + if block.Type != "tool_use" { + kept = append(kept, block) + } + } + msg.Content = kept + continue + } + + // Place consolidated tool_result message immediately after + // the assistant message (position i+1). + targetIdx := i + 1 + + // Build consolidated result message: non-tool_result blocks from the + // first matched user message (if it was at targetIdx) followed by + // all collected tool_results. + consolidated := Message{Role: "user", Content: make([]ContentBlock, 0, len(collectedResults)+len(resultMsgs[0].content))} + if resultMsgs[0].idx == targetIdx && len(resultMsgs[0].content) > 0 { + consolidated.Content = append(consolidated.Content, resultMsgs[0].content...) + } + consolidated.Content = append(consolidated.Content, collectedResults...) + + // Build new message slice. + var newMsgs []Message + newMsgs = append(newMsgs, messages[:targetIdx]...) + newMsgs = append(newMsgs, consolidated) + + // Collect remaining non-empty content from messages between + // targetIdx and scanEnd, excluding the tool_results we moved. + // Skip the first result message if it was already merged at targetIdx. + for j := targetIdx; j < scanEnd && j < len(messages); j++ { + // Determine remaining content for this message. + var remaining []ContentBlock + role := messages[j].Role + isResultMsg := false + for _, rm := range resultMsgs { + if rm.idx == j { + isResultMsg = true + if j == targetIdx { + // Already merged into consolidated above. + remaining = nil + } else { + remaining = rm.content + } + break + } + } + if !isResultMsg { + remaining = messages[j].Content + } + if len(remaining) > 0 { + newMsgs = append(newMsgs, Message{Role: role, Content: remaining}) + } + } + + // Append everything after scanEnd. + if scanEnd < len(messages) { + newMsgs = append(newMsgs, messages[scanEnd:]...) + } + + // Copy newMsgs in-place into messages and truncate. + // copy handles the min(len(messages), len(newMsgs)) case correctly. + prevLen := len(messages) + copy(messages, newMsgs) + messages = messages[:len(newMsgs)] + if len(newMsgs) < prevLen { + i -= prevLen - len(newMsgs) + } + } + return messages +} + // wrapStreamWithTransform lazily captures the model from the first // message_start event in the stream, then applies the provided transform // function to the remaining events. If no transform is set or the first