fix: preserve parallel Claude tool block lifecycle in OpenAI→Claude stream conversion

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 (

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",

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
 }