From 919b466b7202070607cf03caa5ee194ada41b870 Mon Sep 17 00:00:00 2001 From: Junwei Si Date: Tue, 1 Sep 2026 14:40:18 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20OpenAI=E2=86=92Clau?= =?UTF-8?q?de=20=E6=B5=81=E5=BC=8F=E8=BD=AC=E6=8D=A2=E5=9C=A8=E5=B9=B6?= =?UTF-8?q?=E8=A1=8C=E5=B7=A5=E5=85=B7=E8=B0=83=E7=94=A8=E4=B8=8B=E7=9A=84?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E7=B2=98=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复 issue #5483 场景：带 reasoning 的 OpenAI 模型（如 GPT-5.6）流式 /v1/messages + 多 content block（reasoning+text+并行 tool_use）时， Claude SDK 报 InputValidationError: Read was called with input that could not be parsed as JSON。 根因（StreamResponseOpenAI2Claude 为单 block 状态机）： - 上游 tool_calls[].index 从未被读取：同 chunk 内第二个及以后的 tool call 增量直接丢弃 - 第二个 tool call 的参数片段全部并入第一个 tool_use block， 生成 {}{}{} 非法 JSON - None→tools 切换时对未 start 的 block 发幽灵 content_block_stop - 首块即 tool_calls 时提前 return，首块增量丢失 - usage-only 尾 chunk 导致收尾序列静默丢失 修复： - ClaudeConvertInfo 新增 ToolCallBlockIndex（上游 index→block index 映射）与 ActiveBlocks（活跃 block 跟踪） - tools 分支遍历全部增量并按上游 index 路由到各自 content block - Done 终结调用提前，所有活跃 block 统一补发 stop 后再收尾 测试：新增 service/convert_test.go（协议状态机校验器 + 6 用例， 含 2 个回归保护），全部通过；go vet / go build 通过。 --- relay/common/relay_info.go | 5 + service/convert.go | 275 +++++++++++++---------- service/convert_test.go | 448 +++++++++++++++++++++++++++++++++++++ 3 files changed, 610 insertions(+), 118 deletions(-) create mode 100644 service/convert_test.go diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 45fde01..c5d4e96 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -31,6 +31,11 @@ type ClaudeConvertInfo struct { Usage *dto.Usage FinishReason string Done bool + // ToolCallBlockIndex 记录 OpenAI tool_calls[].index 到 Anthropic content block index 的映射， + // 用于把并行工具调用的增量路由到各自的 content block（修复多 block 流式下参数粘连/丢弃） + ToolCallBlockIndex map[int]int + // ActiveBlocks 记录已 start 且未 stop 的 block index，流结束时统一补发 stop，防止遗漏 + ActiveBlocks []int } const ( diff --git a/service/convert.go b/service/convert.go index 593b59d..8b7bc27 100644 --- a/service/convert.go +++ b/service/convert.go @@ -8,6 +8,7 @@ import ( "one-api/dto" "one-api/relay/channel/openrouter" relaycommon "one-api/relay/common" + "sort" "strings" ) @@ -217,8 +218,29 @@ func generateStopBlock(index int) *dto.ClaudeResponse { } } +// addActiveBlock 记录一个已 start 未 stop 的 block index（集合语义，防重复） +func addActiveBlock(convertInfo *relaycommon.ClaudeConvertInfo, index int) { + for _, v := range convertInfo.ActiveBlocks { + if v == index { + return + } + } + convertInfo.ActiveBlocks = append(convertInfo.ActiveBlocks, index) +} + +// removeActiveBlock 将已 stop 的 block index 从活跃集合中移除 +func removeActiveBlock(convertInfo *relaycommon.ClaudeConvertInfo, index int) { + for i, v := range convertInfo.ActiveBlocks { + if v == index { + convertInfo.ActiveBlocks = append(convertInfo.ActiveBlocks[:i], convertInfo.ActiveBlocks[i+1:]...) + return + } + } +} + func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamResponse, info *relaycommon.RelayInfo) []*dto.ClaudeResponse { var claudeResponses []*dto.ClaudeResponse + convertInfo := info.ClaudeConvertInfo if info.SendResponseCount == 1 { msg := &dto.ClaudeMediaMessage{ Id: openAIResponse.Id, @@ -235,32 +257,42 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon Type: "message_start", Message: msg, }) - claudeResponses = append(claudeResponses) - //claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - // Type: "ping", - //}) - if openAIResponse.IsToolCall() { - resp := &dto.ClaudeResponse{ - Type: "content_block_start", - ContentBlock: &dto.ClaudeMediaMessage{ - Id: openAIResponse.GetFirstToolCall().ID, - Type: "tool_use", - Name: openAIResponse.GetFirstToolCall().Function.Name, + // 首个 chunk 的内容（thinking/text/tool_calls）不再在此提前 return， + // 继续走下方通用逻辑，避免首块即 tool_calls 时丢失增量 + } + + // 终结调用（Done，由 handleFinalResponse 触发）： + // 无论最后一个 chunk 的形态（带 finish_reason / usage-only / 空 choices），都必须补齐收尾序列 + if convertInfo.Done { + if len(openAIResponse.Choices) > 0 && openAIResponse.Choices[0].FinishReason != nil && *openAIResponse.Choices[0].FinishReason != "" { + convertInfo.FinishReason = *openAIResponse.Choices[0].FinishReason + } + // 关闭所有活跃 block（按 index 升序），防止 content_block_stop 遗漏 + stops := make([]int, len(convertInfo.ActiveBlocks)) + copy(stops, convertInfo.ActiveBlocks) + sort.Ints(stops) + for _, idx := range stops { + claudeResponses = append(claudeResponses, generateStopBlock(idx)) + } + convertInfo.ActiveBlocks = nil + oaiUsage := convertInfo.Usage + if oaiUsage != nil { + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Type: "message_delta", + Usage: &dto.ClaudeUsage{ + InputTokens: oaiUsage.PromptTokens, + OutputTokens: oaiUsage.CompletionTokens, + CacheCreationInputTokens: oaiUsage.PromptTokensDetails.CachedCreationTokens, + CacheReadInputTokens: oaiUsage.PromptTokensDetails.CachedTokens, }, - } - resp.SetIndex(0) - claudeResponses = append(claudeResponses, resp) - } else { - //resp := &dto.ClaudeResponse{ - // Type: "content_block_start", - // ContentBlock: &dto.ClaudeMediaMessage{ - // Type: "text", - // Text: common.GetPointer[string](""), - // }, - //} - //resp.SetIndex(0) - //claudeResponses = append(claudeResponses, resp) + Delta: &dto.ClaudeMediaMessage{ + StopReason: common.GetPointer[string](stopReasonOpenAI2Claude(convertInfo.FinishReason)), + }, + }) } + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Type: "message_stop", + }) return claudeResponses } @@ -268,112 +300,119 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon // no choices // TODO: handle this case return claudeResponses - } else { - chosenChoice := openAIResponse.Choices[0] - if chosenChoice.FinishReason != nil && *chosenChoice.FinishReason != "" { - // should be done - info.FinishReason = *chosenChoice.FinishReason - return claudeResponses + } + chosenChoice := openAIResponse.Choices[0] + if chosenChoice.FinishReason != nil && *chosenChoice.FinishReason != "" { + // should be done + convertInfo.FinishReason = *chosenChoice.FinishReason + return claudeResponses + } + + if len(chosenChoice.Delta.ToolCalls) > 0 { + // ===== tool_calls 增量：按上游 tool_calls[].index 路由到各自的 content block ===== + if convertInfo.ToolCallBlockIndex == nil { + convertInfo.ToolCallBlockIndex = make(map[int]int) + } + // 从 text/thinking 切换到 tools：先关闭当前文本类 block + if convertInfo.LastMessagesType == relaycommon.LastMessageTypeText || + convertInfo.LastMessagesType == relaycommon.LastMessageTypeThinking { + claudeResponses = append(claudeResponses, generateStopBlock(convertInfo.Index)) + removeActiveBlock(convertInfo, convertInfo.Index) + convertInfo.Index++ } - if info.Done { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - oaiUsage := info.ClaudeConvertInfo.Usage - if oaiUsage != nil { + convertInfo.LastMessagesType = relaycommon.LastMessageTypeTools + for _, toolCall := range chosenChoice.Delta.ToolCalls { + // 上游 index 缺失时按 0 处理（单工具上游常见省略） + oaiIndex := 0 + if toolCall.Index != nil { + oaiIndex = *toolCall.Index + } + blockIndex, ok := convertInfo.ToolCallBlockIndex[oaiIndex] + if !ok { + // 新的 tool call：分配新的 content block（首个 block 沿用当前 Index，后续递增） + if len(convertInfo.ToolCallBlockIndex) > 0 { + convertInfo.Index++ + } + blockIndex = convertInfo.Index + convertInfo.ToolCallBlockIndex[oaiIndex] = blockIndex + addActiveBlock(convertInfo, blockIndex) claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - Type: "message_delta", - Usage: &dto.ClaudeUsage{ - InputTokens: oaiUsage.PromptTokens, - OutputTokens: oaiUsage.CompletionTokens, - CacheCreationInputTokens: oaiUsage.PromptTokensDetails.CachedCreationTokens, - CacheReadInputTokens: oaiUsage.PromptTokensDetails.CachedTokens, + Index: common.GetPointer[int](blockIndex), + Type: "content_block_start", + ContentBlock: &dto.ClaudeMediaMessage{ + Id: toolCall.ID, + Type: "tool_use", + Name: toolCall.Function.Name, + Input: map[string]interface{}{}, }, + }) + } + if toolCall.Function.Arguments != "" { + arguments := toolCall.Function.Arguments + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Index: common.GetPointer[int](blockIndex), + Type: "content_block_delta", Delta: &dto.ClaudeMediaMessage{ - StopReason: common.GetPointer[string](stopReasonOpenAI2Claude(info.FinishReason)), + Type: "input_json_delta", + PartialJson: &arguments, }, }) } + } + return claudeResponses + } + + // ===== text / thinking 增量 ===== + reasoning := chosenChoice.Delta.GetReasoningContent() + textContent := chosenChoice.Delta.GetContentString() + if reasoning == "" && textContent == "" { + return claudeResponses + } + var claudeResponse dto.ClaudeResponse + claudeResponse.Type = "content_block_delta" + if reasoning != "" { + if convertInfo.LastMessagesType != relaycommon.LastMessageTypeThinking { claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - Type: "message_stop", + Index: common.GetPointer[int](convertInfo.Index), + Type: "content_block_start", + ContentBlock: &dto.ClaudeMediaMessage{ + Type: "thinking", + Thinking: "", + }, }) - } else { - var claudeResponse dto.ClaudeResponse - var isEmpty bool - claudeResponse.Type = "content_block_delta" - if len(chosenChoice.Delta.ToolCalls) > 0 { - if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeTools { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ - claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - Index: &info.ClaudeConvertInfo.Index, - Type: "content_block_start", - ContentBlock: &dto.ClaudeMediaMessage{ - Id: openAIResponse.GetFirstToolCall().ID, - Type: "tool_use", - Name: openAIResponse.GetFirstToolCall().Function.Name, - Input: map[string]interface{}{}, - }, - }) - } - info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeTools - // tools delta - claudeResponse.Delta = &dto.ClaudeMediaMessage{ - Type: "input_json_delta", - PartialJson: &chosenChoice.Delta.ToolCalls[0].Function.Arguments, - } - } else { - reasoning := chosenChoice.Delta.GetReasoningContent() - textContent := chosenChoice.Delta.GetContentString() - if reasoning != "" || textContent != "" { - if reasoning != "" { - if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeThinking { - //info.ClaudeConvertInfo.Index++ - claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - Index: &info.ClaudeConvertInfo.Index, - Type: "content_block_start", - ContentBlock: &dto.ClaudeMediaMessage{ - Type: "thinking", - Thinking: "", - }, - }) - } - info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeThinking - // text delta - claudeResponse.Delta = &dto.ClaudeMediaMessage{ - Type: "thinking_delta", - Thinking: reasoning, - } - } else { - if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeText { - if info.LastMessagesType == relaycommon.LastMessageTypeThinking || info.LastMessagesType == relaycommon.LastMessageTypeTools { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ - } - claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ - Index: &info.ClaudeConvertInfo.Index, - Type: "content_block_start", - ContentBlock: &dto.ClaudeMediaMessage{ - Type: "text", - Text: common.GetPointer[string](""), - }, - }) - } - info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeText - // text delta - claudeResponse.Delta = &dto.ClaudeMediaMessage{ - Type: "text_delta", - Text: common.GetPointer[string](textContent), - } - } - } else { - isEmpty = true - } - } - claudeResponse.Index = &info.ClaudeConvertInfo.Index - if !isEmpty { - claudeResponses = append(claudeResponses, &claudeResponse) + addActiveBlock(convertInfo, convertInfo.Index) + } + convertInfo.LastMessagesType = relaycommon.LastMessageTypeThinking + claudeResponse.Index = common.GetPointer[int](convertInfo.Index) + claudeResponse.Delta = &dto.ClaudeMediaMessage{ + Type: "thinking_delta", + Thinking: reasoning, + } + } else { + if convertInfo.LastMessagesType != relaycommon.LastMessageTypeText { + if convertInfo.LastMessagesType == relaycommon.LastMessageTypeThinking || convertInfo.LastMessagesType == relaycommon.LastMessageTypeTools { + claudeResponses = append(claudeResponses, generateStopBlock(convertInfo.Index)) + removeActiveBlock(convertInfo, convertInfo.Index) + convertInfo.Index++ } + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Index: common.GetPointer[int](convertInfo.Index), + Type: "content_block_start", + ContentBlock: &dto.ClaudeMediaMessage{ + Type: "text", + Text: common.GetPointer[string](""), + }, + }) + addActiveBlock(convertInfo, convertInfo.Index) + } + convertInfo.LastMessagesType = relaycommon.LastMessageTypeText + claudeResponse.Index = common.GetPointer[int](convertInfo.Index) + claudeResponse.Delta = &dto.ClaudeMediaMessage{ + Type: "text_delta", + Text: common.GetPointer[string](textContent), } } + claudeResponses = append(claudeResponses, &claudeResponse) return claudeResponses } diff --git a/service/convert_test.go b/service/convert_test.go new file mode 100644 index 0000000..cabd5f8 --- /dev/null +++ b/service/convert_test.go @@ -0,0 +1,448 @@ +package service + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + + "one-api/dto" + relaycommon "one-api/relay/common" +) + +// ============ 测试驱动器：忠实模拟 OaiStreamHandler 的调用契约 ============ +// +// 真实链路（relay/channel/openai/relay-openai.go）： +// - 前 n-1 个 data chunk 经 handleStreamFormat 处理：先 SendResponseCount++ 再调 convert +// - 最后 1 个 data chunk 经 handleFinalResponse 处理：先置 ClaudeConvertInfo.Done=true 再调 convert（不递增计数） +// +// 参考 issue #5483：GPT-5.6 流式 /v1/messages + 多 content block（reasoning+text+并行 tool_use）场景。 + +func streamChunk(delta string) string { + return fmt.Sprintf(`{"id":"chatcmpl-test","object":"chat.completion.chunk","model":"gpt-5.6","choices":[{"index":0,"delta":%s}]}`, delta) +} + +func streamChunkFinish(reason string) string { + return fmt.Sprintf(`{"id":"chatcmpl-test","object":"chat.completion.chunk","model":"gpt-5.6","choices":[{"index":0,"delta":{},"finish_reason":"%s"}]}`, reason) +} + +// 尾部 chunk 带空 choices 条目 + usage（OpenAI 兼容上游常见流尾形态） +func streamChunkTailWithUsage() string { + return `{"id":"chatcmpl-test","object":"chat.completion.chunk","model":"gpt-5.6","choices":[{"index":0,"delta":{}}],"usage":{"prompt_tokens":10,"completion_tokens":100,"total_tokens":110}}` +} + +// 尾部 chunk 为 usage-only（choices 为空数组，OpenAI 官方流尾形态） +func streamChunkUsageOnlyTail() string { + return `{"id":"chatcmpl-test","object":"chat.completion.chunk","model":"gpt-5.6","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":100,"total_tokens":110}}` +} + +// driveClaudeStream 按真实调用契约驱动流式转换，返回全部输出事件 +func driveClaudeStream(t *testing.T, chunks []string) []*dto.ClaudeResponse { + t.Helper() + info := &relaycommon.RelayInfo{ + PromptTokens: 10, + ClaudeConvertInfo: &relaycommon.ClaudeConvertInfo{ + LastMessagesType: relaycommon.LastMessageTypeNone, + }, + } + var events []*dto.ClaudeResponse + for i, chunk := range chunks { + var streamResponse dto.ChatCompletionsStreamResponse + if err := json.Unmarshal([]byte(chunk), &streamResponse); err != nil { + t.Fatalf("解析测试 chunk 失败: %v\nchunk: %s", err, chunk) + } + // 模拟 handleClaudeFormat/handleFinalResponse 的 usage 提取 + if streamResponse.Usage != nil { + info.ClaudeConvertInfo.Usage = streamResponse.Usage + } + if i == len(chunks)-1 { + // 最后一个 chunk：模拟 handleFinalResponse + info.ClaudeConvertInfo.Done = true + } else { + // 其余 chunk：模拟 handleStreamFormat + info.SendResponseCount++ + } + // 模拟 helper.ClaudeData 的即时序列化：每个事件返回后立即做值快照， + // 避免生产代码中 Index 指针共享导致的"事后读取读到最终值"假象 + for _, resp := range StreamResponseOpenAI2Claude(&streamResponse, info) { + snapshot := *resp + if resp.Index != nil { + idx := *resp.Index + snapshot.Index = &idx + } + events = append(events, &snapshot) + } + } + return events +} + +// ============ 协议状态机校验器 ============ + +type blockState struct { + blockType string // text / thinking / tool_use + id string + name string + json strings.Builder // input_json_delta 拼接 + text strings.Builder // text_delta 拼接 + thinking strings.Builder // thinking_delta 拼接 + stopped bool +} + +// validateClaudeStream 按 Anthropic /v1/messages 流式协议校验事件序列合法性， +// 返回按 block index 聚合的内容摘要供语义断言。 +// 规则：block index 从 0 连续分配；delta/stop 只能指向已 start 且未 stop 的 block； +// message_delta 前所有 block 必须已 stop；流必须以 message_stop 收尾。 +func validateClaudeStream(t *testing.T, events []*dto.ClaudeResponse) map[int]*blockState { + t.Helper() + blocks := map[int]*blockState{} + var ( + sawMessageStart bool + sawMessageDelta bool + sawMessageStop bool + nextIndex int + ) + for i, ev := range events { + switch ev.Type { + case "message_start": + if sawMessageStart { + t.Errorf("事件 %d: message_start 出现多次", i) + } + sawMessageStart = true + case "content_block_start": + if ev.ContentBlock == nil { + t.Fatalf("事件 %d: content_block_start 缺少 content_block", i) + } + idx := ev.GetIndex() + if _, ok := blocks[idx]; ok { + t.Fatalf("事件 %d: content_block_start 重复 start index %d", i, idx) + } + if idx != nextIndex { + t.Fatalf("事件 %d: content_block_start index %d 不连续（期望 %d）", i, idx, nextIndex) + } + nextIndex++ + blocks[idx] = &blockState{ + blockType: ev.ContentBlock.Type, + id: ev.ContentBlock.Id, + name: ev.ContentBlock.Name, + } + case "content_block_delta": + idx := ev.GetIndex() + bs, ok := blocks[idx] + if !ok { + t.Fatalf("事件 %d: content_block_delta 指向未 start 的 block %d", i, idx) + } + if bs.stopped { + t.Fatalf("事件 %d: content_block_delta 指向已 stop 的 block %d", i, idx) + } + if ev.Delta == nil { + t.Fatalf("事件 %d: content_block_delta 缺少 delta", i) + } + switch ev.Delta.Type { + case "text_delta": + if bs.blockType != "text" { + t.Errorf("事件 %d: text_delta 发给了 %s block %d", i, bs.blockType, idx) + } + bs.text.WriteString(ev.Delta.GetText()) + case "thinking_delta": + if bs.blockType != "thinking" { + t.Errorf("事件 %d: thinking_delta 发给了 %s block %d", i, bs.blockType, idx) + } + bs.thinking.WriteString(ev.Delta.Thinking) + case "input_json_delta": + if bs.blockType != "tool_use" { + t.Errorf("事件 %d: input_json_delta 发给了 %s block %d", i, bs.blockType, idx) + } + if ev.Delta.PartialJson != nil { + bs.json.WriteString(*ev.Delta.PartialJson) + } + default: + t.Errorf("事件 %d: 未知 delta 类型 %q", i, ev.Delta.Type) + } + case "content_block_stop": + idx := ev.GetIndex() + bs, ok := blocks[idx] + if !ok { + t.Fatalf("事件 %d: content_block_stop 指向未 start 的 block %d —— 幽灵 stop", i, idx) + } + if bs.stopped { + t.Fatalf("事件 %d: content_block_stop 重复 stop block %d", i, idx) + } + bs.stopped = true + case "message_delta": + for idx, bs := range blocks { + if !bs.stopped { + t.Errorf("message_delta 前存在未 stop 的 block %d（%s）", idx, bs.blockType) + } + } + sawMessageDelta = true + case "message_stop": + if !sawMessageDelta { + t.Errorf("事件 %d: message_stop 前缺少 message_delta", i) + } + sawMessageStop = true + case "ping": + // 心跳事件，允许 + default: + t.Errorf("事件 %d: 未知事件类型 %q", i, ev.Type) + } + } + if !sawMessageStart { + t.Errorf("流缺少 message_start") + } + if !sawMessageStop { + t.Errorf("流未正常结束：缺少 message_stop") + } + return blocks +} + +// ============ 断言辅助 ============ + +func summarizeBlocks(blocks map[int]*blockState) string { + var sb strings.Builder + for i := 0; ; i++ { + bs, ok := blocks[i] + if !ok { + break + } + fmt.Fprintf(&sb, "[%d]=%s ", i, bs.blockType) + } + return sb.String() +} + +func assertBlock(t *testing.T, blocks map[int]*blockState, idx int, wantType, wantId, wantName string) { + t.Helper() + bs, ok := blocks[idx] + if !ok { + t.Fatalf("缺少 block %d，实际 blocks: %s", idx, summarizeBlocks(blocks)) + } + if bs.blockType != wantType { + t.Errorf("block %d 类型 = %q，期望 %q", idx, bs.blockType, wantType) + } + if wantId != "" && bs.id != wantId { + t.Errorf("block %d id = %q，期望 %q", idx, bs.id, wantId) + } + if wantName != "" && bs.name != wantName { + t.Errorf("block %d name = %q，期望 %q", idx, bs.name, wantName) + } +} + +// assertToolJson 断言某 tool_use block 的 input_json_delta 拼接结果： +// 必须精确等于期望值，且是合法 JSON 对象（否则 Claude SDK 报 InputValidationError） +func assertToolJson(t *testing.T, blocks map[int]*blockState, idx int, want string) { + t.Helper() + bs, ok := blocks[idx] + if !ok { + t.Fatalf("缺少 block %d，实际 blocks: %s", idx, summarizeBlocks(blocks)) + } + got := bs.json.String() + if got != want { + t.Errorf("block %d 的 input_json_delta 拼接 = %q，期望 %q（并行工具参数粘连？）", idx, got, want) + } + var obj map[string]any + if err := json.Unmarshal([]byte(got), &obj); err != nil { + t.Errorf("block %d 的 input_json_delta 拼接不是合法 JSON: %q —— Claude SDK 将报 InputValidationError", idx, got) + } +} + +func findMessageDelta(events []*dto.ClaudeResponse) *dto.ClaudeResponse { + for i := range events { + if events[i].Type == "message_delta" { + return events[i] + } + } + return nil +} + +func assertStopReason(t *testing.T, events []*dto.ClaudeResponse, want string) { + t.Helper() + md := findMessageDelta(events) + if md == nil { + t.Fatalf("缺少 message_delta 事件") + } + if md.Delta == nil || md.Delta.StopReason == nil { + t.Fatalf("message_delta 缺少 delta.stop_reason") + } + if *md.Delta.StopReason != want { + t.Errorf("stop_reason = %q，期望 %q", *md.Delta.StopReason, want) + } +} + +// ============ 测试用例 ============ + +// 核心场景（issue #5483）：reasoning + text + 并行双 tool_use 的多 block 流式。 +// 修复前：第二个 tool call 的增量被并入第一个 tool block（JSON 粘连）， +// Claude SDK 报 InputValidationError: Read was called with input that could not be parsed as JSON。 +func TestStreamResponseOpenAI2Claude_ParallelToolCalls(t *testing.T) { + chunks := []string{ + streamChunk(`{"role":"assistant","content":""}`), + streamChunk(`{"reasoning_content":"先思考"}`), + streamChunk(`{"content":"需要读两个文件"}`), + // 并行双 tool：首个 tool chunk 的数组里带两个 tool call（各带 id/name/index） + streamChunk(`{"tool_calls":[ + {"index":0,"id":"call_a","type":"function","function":{"name":"read_file","arguments":""}}, + {"index":1,"id":"call_b","type":"function","function":{"name":"write_file","arguments":""}}]}`), + // 交错参数增量：上游按 index 路由片段 + streamChunk(`{"tool_calls":[{"index":0,"function":{"arguments":"{\"path\":"}}]}`), + streamChunk(`{"tool_calls":[{"index":1,"function":{"arguments":"{\"path\":"}}]}`), + streamChunk(`{"tool_calls":[{"index":0,"function":{"arguments":"\"a.txt\"}"}}]}`), + streamChunk(`{"tool_calls":[{"index":1,"function":{"arguments":"\"b.txt\"}"}}]}`), + streamChunkFinish("tool_calls"), + streamChunkTailWithUsage(), + } + events := driveClaudeStream(t, chunks) + blocks := validateClaudeStream(t, events) + + // 期望 4 个 block：thinking(0) + text(1) + tool_use(2) + tool_use(3) + if len(blocks) != 4 { + t.Fatalf("期望 4 个 content block，实际 %d 个: %s", len(blocks), summarizeBlocks(blocks)) + } + assertBlock(t, blocks, 0, "thinking", "", "") + assertBlock(t, blocks, 1, "text", "", "") + assertBlock(t, blocks, 2, "tool_use", "call_a", "read_file") + assertBlock(t, blocks, 3, "tool_use", "call_b", "write_file") + + // 两个 tool 的参数各自完整、互不粘连 + assertToolJson(t, blocks, 2, `{"path":"a.txt"}`) + assertToolJson(t, blocks, 3, `{"path":"b.txt"}`) + + // 文本与思考内容不串块 + if got := blocks[1].text.String(); got != "需要读两个文件" { + t.Errorf("text block 内容 = %q，期望 %q", got, "需要读两个文件") + } + if got := blocks[0].thinking.String(); got != "先思考" { + t.Errorf("thinking block 内容 = %q，期望 %q", got, "先思考") + } + + assertStopReason(t, events, "tool_use") +} + +// 回归：纯文本流式（修复前后行为应一致） +func TestStreamResponseOpenAI2Claude_TextOnly(t *testing.T) { + chunks := []string{ + streamChunk(`{"role":"assistant","content":""}`), + streamChunk(`{"content":"你好"}`), + streamChunk(`{"content":"世界"}`), + streamChunkFinish("stop"), + streamChunkTailWithUsage(), + } + events := driveClaudeStream(t, chunks) + blocks := validateClaudeStream(t, events) + + if len(blocks) != 1 || blocks[0].blockType != "text" { + t.Fatalf("期望单个 text block，实际: %s", summarizeBlocks(blocks)) + } + if got := blocks[0].text.String(); got != "你好世界" { + t.Errorf("text block 内容 = %q，期望 %q", got, "你好世界") + } + assertStopReason(t, events, "end_turn") + + // message_start 的 usage 来自 PromptTokens + if len(events) == 0 || events[0].Type != "message_start" || events[0].Message == nil { + t.Fatalf("首个事件应为 message_start") + } + if events[0].Message.Usage == nil || events[0].Message.Usage.InputTokens != 10 { + t.Errorf("message_start 的 input_tokens 期望 10") + } + // message_delta 的 usage 来自尾部 usage chunk + md := findMessageDelta(events) + if md == nil || md.Usage == nil || md.Usage.OutputTokens != 100 { + t.Errorf("message_delta 的 output_tokens 期望 100") + } +} + +// 单 tool 流式：不允许出现"幽灵 stop"（对未 start 过的 block 发 content_block_stop）， +// 首个 block 的 index 应为 0。 +func TestStreamResponseOpenAI2Claude_SingleToolNoGhostStop(t *testing.T) { + chunks := []string{ + streamChunk(`{"role":"assistant","content":""}`), + streamChunk(`{"tool_calls":[{"index":0,"id":"call_a","type":"function","function":{"name":"read_file","arguments":""}}]}`), + streamChunk(`{"tool_calls":[{"index":0,"function":{"arguments":"{\"path\":\"a.txt\"}"}}]}`), + streamChunkFinish("tool_calls"), + streamChunkTailWithUsage(), + } + events := driveClaudeStream(t, chunks) + blocks := validateClaudeStream(t, events) + + if len(blocks) != 1 { + t.Fatalf("期望单个 tool_use block，实际: %s", summarizeBlocks(blocks)) + } + assertBlock(t, blocks, 0, "tool_use", "call_a", "read_file") + assertToolJson(t, blocks, 0, `{"path":"a.txt"}`) +} + +// 回归：reasoning → text → 单 tool（GPT-5.6 常见序列，修复前后行为应一致） +func TestStreamResponseOpenAI2Claude_ReasoningTextSingleTool(t *testing.T) { + chunks := []string{ + streamChunk(`{"role":"assistant","content":""}`), + streamChunk(`{"reasoning_content":"思考一下"}`), + streamChunk(`{"content":"读取文件"}`), + streamChunk(`{"tool_calls":[{"index":0,"id":"call_a","type":"function","function":{"name":"read_file","arguments":""}}]}`), + streamChunk(`{"tool_calls":[{"index":0,"function":{"arguments":"{\"path\":\"a.txt\"}"}}]}`), + streamChunkFinish("tool_calls"), + streamChunkTailWithUsage(), + } + events := driveClaudeStream(t, chunks) + blocks := validateClaudeStream(t, events) + + if len(blocks) != 3 { + t.Fatalf("期望 3 个 block（thinking/text/tool_use），实际: %s", summarizeBlocks(blocks)) + } + assertBlock(t, blocks, 0, "thinking", "", "") + assertBlock(t, blocks, 1, "text", "", "") + assertBlock(t, blocks, 2, "tool_use", "call_a", "read_file") + assertToolJson(t, blocks, 2, `{"path":"a.txt"}`) + if got := blocks[0].thinking.String(); got != "思考一下" { + t.Errorf("thinking 内容 = %q", got) + } + if got := blocks[1].text.String(); got != "读取文件" { + t.Errorf("text 内容 = %q", got) + } + assertStopReason(t, events, "tool_use") +} + +// 首个 chunk 即并行双 tool（无 role/reasoning/text 前导块）： +// message_start 后应正确 start 两个 tool_use block，首个 chunk 的 arguments 不丢失 +func TestStreamResponseOpenAI2Claude_FirstChunkParallelTools(t *testing.T) { + chunks := []string{ + streamChunk(`{"tool_calls":[ + {"index":0,"id":"call_a","type":"function","function":{"name":"read_file","arguments":"{\"path\":"}}, + {"index":1,"id":"call_b","type":"function","function":{"name":"write_file","arguments":"{\"path\":"}}]}`), + streamChunk(`{"tool_calls":[{"index":0,"function":{"arguments":"\"a.txt\"}"}}]}`), + streamChunk(`{"tool_calls":[{"index":1,"function":{"arguments":"\"b.txt\"}"}}]}`), + streamChunkFinish("tool_calls"), + streamChunkTailWithUsage(), + } + events := driveClaudeStream(t, chunks) + blocks := validateClaudeStream(t, events) + + if len(blocks) != 2 { + t.Fatalf("期望 2 个 tool_use block，实际: %s", summarizeBlocks(blocks)) + } + assertBlock(t, blocks, 0, "tool_use", "call_a", "read_file") + assertBlock(t, blocks, 1, "tool_use", "call_b", "write_file") + assertToolJson(t, blocks, 0, `{"path":"a.txt"}`) + assertToolJson(t, blocks, 1, `{"path":"b.txt"}`) + assertStopReason(t, events, "tool_use") +} + +// 尾部 chunk 为 usage-only（choices 为空数组，OpenAI 官方流尾形态）时， +// 流仍必须完整收尾（所有 block stop + message_delta + message_stop）， +// 否则 Claude SDK 报流意外中断。 +func TestStreamResponseOpenAI2Claude_UsageOnlyTail(t *testing.T) { + chunks := []string{ + streamChunk(`{"role":"assistant","content":""}`), + streamChunk(`{"content":"你好"}`), + streamChunkFinish("stop"), + streamChunkUsageOnlyTail(), + } + events := driveClaudeStream(t, chunks) + blocks := validateClaudeStream(t, events) + + if len(blocks) != 1 || blocks[0].blockType != "text" { + t.Fatalf("期望单个 text block，实际: %s", summarizeBlocks(blocks)) + } + if got := blocks[0].text.String(); got != "你好" { + t.Errorf("text block 内容 = %q", got) + } + assertStopReason(t, events, "end_turn") +}