diff --git a/relaykit/relayconvert/convmeta/meta.go b/relaykit/relayconvert/convmeta/meta.go index 68848af9715a..dddbadff9eff 100644 --- a/relaykit/relayconvert/convmeta/meta.go +++ b/relaykit/relayconvert/convmeta/meta.go @@ -60,6 +60,10 @@ type ClaudeConvertInfo struct { ToolCallBaseIndex int ToolCallMaxIndexOffset int + // ToolCallOpenIndexes tracks started-but-not-stopped parallel tool_use + // block indexes; nil outside a tools run. Guards delta/stop so they never + // target an index without an active content_block_start (#4389). + ToolCallOpenIndexes map[int]bool } const ( diff --git a/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp.go b/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp.go index 554e70a9d5e9..91cef03220b4 100644 --- a/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp.go +++ b/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp.go @@ -1,6 +1,7 @@ package oaichat import ( + "sort" "strings" "github.com/QuantumNous/new-api/relaykit/dto" @@ -10,6 +11,12 @@ import ( "github.com/samber/lo" ) +// maxToolCallBlockIndex bounds upstream-provided tool_call.index values so a +// malicious or malformed huge index cannot drive a non-terminating stop scan +// or pollute state.Index with an absurd content block index. Real tool_call +// indexes are 0-based and small; this is a defensive ceiling only. +const maxToolCallBlockIndex = 1024 + func generateStopBlock(index int) *dto.ClaudeResponse { return &dto.ClaudeResponse{ Type: "content_block_stop", @@ -25,9 +32,21 @@ func stopOpenBlocks(state *convmeta.ClaudeConvertInfo) []*dto.ClaudeResponse { case convmeta.LastMessageTypeText, convmeta.LastMessageTypeThinking: return []*dto.ClaudeResponse{generateStopBlock(state.Index)} case convmeta.LastMessageTypeTools: - responses := make([]*dto.ClaudeResponse, 0, state.ToolCallMaxIndexOffset+1) - for offset := 0; offset <= state.ToolCallMaxIndexOffset; offset++ { - responses = append(responses, generateStopBlock(state.ToolCallBaseIndex+offset)) + // Iterate the tracked open indexes directly rather than scanning + // 0..ToolCallMaxIndexOffset: O(active blocks) instead of O(max index), + // immune to sparse/huge indexes, and every started block is stopped + // (negative indexes that slipped past validation are still tracked here). + if len(state.ToolCallOpenIndexes) == 0 { + return nil + } + indexes := make([]int, 0, len(state.ToolCallOpenIndexes)) + for idx := range state.ToolCallOpenIndexes { + indexes = append(indexes, idx) + } + sort.Ints(indexes) + responses := make([]*dto.ClaudeResponse, 0, len(indexes)) + for _, idx := range indexes { + responses = append(responses, generateStopBlock(idx)) } return responses default: @@ -125,6 +144,7 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon state.Index = state.ToolCallBaseIndex + state.ToolCallMaxIndexOffset + 1 state.ToolCallBaseIndex = 0 state.ToolCallMaxIndexOffset = 0 + state.ToolCallOpenIndexes = nil default: state.Index++ } @@ -153,39 +173,54 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon state.LastMessagesType = convmeta.LastMessageTypeTools state.ToolCallBaseIndex = 0 state.ToolCallMaxIndexOffset = 0 - var toolCall dto.ToolCallResponse - if len(openAIResponse.Choices) > 0 && len(openAIResponse.Choices[0].Delta.ToolCalls) > 0 { - toolCall = openAIResponse.Choices[0].Delta.ToolCalls[0] - } else { - first := openAIResponse.GetFirstToolCall() - if first != nil { - toolCall = *first - } else { - toolCall = dto.ToolCallResponse{} + state.ToolCallOpenIndexes = make(map[int]bool) + toolCalls := openAIResponse.Choices[0].Delta.ToolCalls + if len(toolCalls) == 0 { + if first := openAIResponse.GetFirstToolCall(); first != nil { + toolCalls = []dto.ToolCallResponse{*first} } } - resp := &dto.ClaudeResponse{ - Type: "content_block_start", - ContentBlock: &dto.ClaudeMediaMessage{ - Id: toolCall.ID, - Type: "tool_use", - Name: toolCall.Function.Name, - Input: map[string]interface{}{}, - }, - } - resp.SetIndex(0) - claudeResponses = append(claudeResponses, resp) - // 首块包含工具 delta，则追加 input_json_delta - if toolCall.Function.Arguments != "" { - idx := 0 - claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - Index: &idx, - Type: "content_block_delta", - Delta: &dto.ClaudeMediaMessage{ - Type: "input_json_delta", - PartialJson: &toolCall.Function.Arguments, - }, - }) + for i, toolCall := range toolCalls { + offset := i + if toolCall.Index != nil { + offset = *toolCall.Index + } + if offset < 0 || offset > maxToolCallBlockIndex { + // reject malformed upstream indexes: a negative or huge index + // would emit an invalid Claude block index and, pre-fix, drove + // a non-terminating stop scan. Skip the tool call entirely. + continue + } + if offset > state.ToolCallMaxIndexOffset { + state.ToolCallMaxIndexOffset = offset + } + idx := offset + // start only when the block carries a name and is not already open, + // mirroring the later-chunk path; a replayed name must not emit a + // second content_block_start for an open index. + if toolCall.Function.Name != "" && !state.ToolCallOpenIndexes[idx] { + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Index: &idx, + Type: "content_block_start", + ContentBlock: &dto.ClaudeMediaMessage{ + Id: toolCall.ID, + Type: "tool_use", + Name: toolCall.Function.Name, + Input: map[string]interface{}{}, + }, + }) + state.ToolCallOpenIndexes[idx] = true + } + if toolCall.Function.Arguments != "" && state.ToolCallOpenIndexes[idx] { + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Index: &idx, + Type: "content_block_delta", + Delta: &dto.ClaudeMediaMessage{ + Type: "input_json_delta", + PartialJson: &toolCall.Function.Arguments, + }, + }) + } } } else { @@ -318,6 +353,7 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon stopOpenBlocksAndAdvance() state.ToolCallBaseIndex = state.Index state.ToolCallMaxIndexOffset = 0 + state.ToolCallOpenIndexes = make(map[int]bool) } state.LastMessagesType = convmeta.LastMessageTypeTools base := state.ToolCallBaseIndex @@ -330,13 +366,20 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon } else { offset = i } + if offset < 0 || offset > maxToolCallBlockIndex { + // reject malformed upstream indexes (see maxToolCallBlockIndex). + continue + } if offset > maxOffset { maxOffset = offset } blockIndex := base + offset idx := blockIndex - if toolCall.Function.Name != "" { + // start only when a name is present and the index is not already + // open; providers that echo the full tool_call each delta would + // otherwise emit a duplicate content_block_start. + if toolCall.Function.Name != "" && !state.ToolCallOpenIndexes[idx] { claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ Index: &idx, Type: "content_block_start", @@ -347,9 +390,13 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon Input: map[string]interface{}{}, }, }) + state.ToolCallOpenIndexes[idx] = true } - if len(toolCall.Function.Arguments) > 0 { + // guard the ghost delta: when args arrive packed in the final + // chunk (e.g. GLM-5.2) after a sibling block was stopped, a + // delta here targets a closed/never-started block (#4389) + if len(toolCall.Function.Arguments) > 0 && state.ToolCallOpenIndexes[idx] { claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ Index: &idx, Type: "content_block_delta", diff --git a/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp_test.go b/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp_test.go index 1ccdd00b2562..8cea5ef51ed0 100644 --- a/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp_test.go +++ b/relaykit/relayconvert/internal/oai_chat/to_claude_messages_resp_test.go @@ -214,6 +214,126 @@ func TestNormalizeCacheCreationSplit(t *testing.T) { assert.Equal(t, 1, cache1h) } +// TestStreamResponseOpenAI2ClaudeParallelToolCallsHaveValidBlockLifecycle +// drives two parallel tool_use blocks (e.g. GLM-5.2 packing multiple tool +// calls per chunk) through the OpenAI→Claude stream converter and asserts the +// Anthropic SSE state machine stays valid: every content_block_delta/stop +// targets an actively-open block index, no block starts twice, and every +// started block is stopped (#4389). +func TestStreamResponseOpenAI2ClaudeParallelToolCallsHaveValidBlockLifecycle(t *testing.T) { + info := &convmeta.Values{ + ClaudeConvertInfo: &convmeta.ClaudeConvertInfo{}, + } + + info.SendResponseCount = 1 + events := StreamResponseOpenAI2Claude(&dto.ChatCompletionsStreamResponse{ + Id: "chatcmpl_1", Model: "glm", + Choices: []dto.ChatCompletionsStreamResponseChoice{{ + Delta: dto.ChatCompletionsStreamResponseChoiceDelta{ToolCalls: []dto.ToolCallResponse{ + {Index: ptr(0), ID: "call_weather", Function: dto.FunctionResponse{Name: "get_weather"}}, + {Index: ptr(1), ID: "call_time", Function: dto.FunctionResponse{Name: "get_time"}}, + }}, + }}, + }, info) + + info.SendResponseCount = 2 + events = append(events, StreamResponseOpenAI2Claude(&dto.ChatCompletionsStreamResponse{ + Choices: []dto.ChatCompletionsStreamResponseChoice{{ + Delta: dto.ChatCompletionsStreamResponseChoiceDelta{ToolCalls: []dto.ToolCallResponse{ + {Index: ptr(0), Function: dto.FunctionResponse{Arguments: `{"city":"Tokyo"}`}}, + {Index: ptr(1), Function: dto.FunctionResponse{Arguments: `{}`}}, + }}, + }}, + }, info)...) + + info.SendResponseCount = 3 + finishReason := "tool_calls" + events = append(events, StreamResponseOpenAI2Claude(&dto.ChatCompletionsStreamResponse{ + Choices: []dto.ChatCompletionsStreamResponseChoice{{FinishReason: &finishReason}}, + Usage: &dto.Usage{}, + }, info)...) + + started := map[int]bool{} + stopped := map[int]bool{} + // capture argument payloads by block index so a converter that drops deltas + // (not just reorders them) still fails the test. + deltas := map[int][]string{} + for _, event := range events { + if event.Index == nil { + continue + } + idx := *event.Index + switch event.Type { + case "content_block_start": + require.False(t, started[idx], "block %d started twice", idx) + started[idx] = true + case "content_block_delta": + assert.True(t, started[idx], "block %d received delta before start", idx) + assert.False(t, stopped[idx], "block %d received delta after stop", idx) + if event.Delta != nil && event.Delta.PartialJson != nil { + deltas[idx] = append(deltas[idx], *event.Delta.PartialJson) + } + case "content_block_stop": + assert.True(t, started[idx], "block %d stopped before start", idx) + require.False(t, stopped[idx], "block %d stopped twice", idx) + stopped[idx] = true + } + } + + assert.Equal(t, map[int]bool{0: true, 1: true}, started) + assert.Equal(t, started, stopped) + assert.Equal(t, []string{`{"city":"Tokyo"}`}, deltas[0], "block 0 must deliver its argument payload") + assert.Equal(t, []string{`{}`}, deltas[1], "block 1 must deliver its argument payload") +} + +// TestStreamResponseOpenAI2ClaudeReplayedToolNameDoesNotDuplicateStart covers +// providers that echo the full tool_call (id+name) in every delta instead of +// streaming incremental fragments: a replayed name for an already-open index +// must not emit a second content_block_start. +func TestStreamResponseOpenAI2ClaudeReplayedToolNameDoesNotDuplicateStart(t *testing.T) { + info := &convmeta.Values{ + ClaudeConvertInfo: &convmeta.ClaudeConvertInfo{}, + } + + info.SendResponseCount = 1 + first := StreamResponseOpenAI2Claude(&dto.ChatCompletionsStreamResponse{ + Id: "chatcmpl_1", Model: "glm", + Choices: []dto.ChatCompletionsStreamResponseChoice{{ + Delta: dto.ChatCompletionsStreamResponseChoiceDelta{ToolCalls: []dto.ToolCallResponse{ + {Index: ptr(0), ID: "call_weather", Function: dto.FunctionResponse{Name: "get_weather"}}, + }}, + }}, + }, info) + + info.SendResponseCount = 2 + // upstream re-echoes name+id alongside an arguments fragment + second := StreamResponseOpenAI2Claude(&dto.ChatCompletionsStreamResponse{ + Choices: []dto.ChatCompletionsStreamResponseChoice{{ + Delta: dto.ChatCompletionsStreamResponseChoiceDelta{ToolCalls: []dto.ToolCallResponse{ + {Index: ptr(0), ID: "call_weather", Function: dto.FunctionResponse{Name: "get_weather", Arguments: `{"city":"Tokyo"}`}}, + }}, + }}, + }, info) + + info.SendResponseCount = 3 + finishReason := "tool_calls" + third := StreamResponseOpenAI2Claude(&dto.ChatCompletionsStreamResponse{ + Choices: []dto.ChatCompletionsStreamResponseChoice{{FinishReason: &finishReason}}, + Usage: &dto.Usage{}, + }, info) + + // collect every content_block_start index; a replayed name must not start a + // new block at any index (e.g. a spurious index 1), so assert the exact set. + var startIndexes []int + for _, event := range append(append(first, second...), third...) { + if event.Type != "content_block_start" || event.Index == nil { + continue + } + startIndexes = append(startIndexes, *event.Index) + } + assert.Equal(t, []int{0}, startIndexes, "only block 0 may start despite replayed name") +} + func ptr[T any](value T) *T { return &value }