diff --git a/internal/translator/codex/claude/codex_claude_parallel_function_calls_test.go b/internal/translator/codex/claude/codex_claude_parallel_function_calls_test.go new file mode 100644 index 000000000..b92fd52a8 --- /dev/null +++ b/internal/translator/codex/claude/codex_claude_parallel_function_calls_test.go @@ -0,0 +1,305 @@ +package claude + +import ( + "context" + "strings" + "testing" + + "github.com/tidwall/gjson" +) + +type codexClaudeContentBlock struct { + Index int64 + Type string + ID string + Name string + Text string + Arguments string +} + +func translateCodexClaudeChunks(t *testing.T, chunks [][]byte) [][]byte { + t.Helper() + + originalRequest := []byte(`{"stream":true,"tools":[{"name":"Read"}]}`) + var state any + var outputs [][]byte + for _, chunk := range chunks { + outputs = append(outputs, ConvertCodexResponseToClaude(context.Background(), "gpt-5", originalRequest, nil, chunk, &state)...) + } + return outputs +} + +func assertCodexClaudeContentBlockLifecycle(t *testing.T, outputs [][]byte) []*codexClaudeContentBlock { + t.Helper() + + open := make(map[int64]*codexClaudeContentBlock) + started := make(map[int64]struct{}) + blocks := make([]*codexClaudeContentBlock, 0) + messageState := 0 + for _, output := range outputs { + for _, line := range strings.Split(string(output), "\n") { + if !strings.HasPrefix(line, "data: ") { + continue + } + event := gjson.Parse(strings.TrimPrefix(line, "data: ")) + if messageState == 2 { + t.Fatalf("event emitted after message_stop: %s", event.Raw) + } + index := event.Get("index").Int() + switch event.Get("type").String() { + case "content_block_start": + if messageState != 0 { + t.Fatalf("content block started after message terminal events: %s", event.Raw) + } + if len(open) != 0 { + t.Fatalf("content block start emitted while another block remains open: %v", open) + } + if _, exists := started[index]; exists { + t.Fatalf("content block index %d was reused", index) + } + block := &codexClaudeContentBlock{ + Index: index, + Type: event.Get("content_block.type").String(), + ID: event.Get("content_block.id").String(), + Name: event.Get("content_block.name").String(), + } + open[index] = block + started[index] = struct{}{} + blocks = append(blocks, block) + case "content_block_delta": + block := open[index] + if block == nil { + t.Fatalf("content block delta targets unopened index %d", index) + } + switch event.Get("delta.type").String() { + case "input_json_delta": + block.Arguments += event.Get("delta.partial_json").String() + case "text_delta": + block.Text += event.Get("delta.text").String() + } + case "content_block_stop": + if open[index] == nil { + t.Fatalf("content block stop targets unopened index %d", index) + } + delete(open, index) + case "message_delta": + if len(open) != 0 { + t.Fatalf("message_delta emitted while content blocks remain open: %v", open) + } + if messageState != 0 { + t.Fatalf("duplicate or out-of-order message_delta: %s", event.Raw) + } + messageState = 1 + case "message_stop": + if len(open) != 0 { + t.Fatalf("message_stop emitted while content blocks remain open: %v", open) + } + if messageState != 1 { + t.Fatalf("message_stop emitted before message_delta: %s", event.Raw) + } + messageState = 2 + } + } + } + if len(open) != 0 { + t.Fatalf("content blocks remain open: %v", open) + } + return blocks +} + +func assertParallelCodexClaudeToolCalls(t *testing.T, blocks []*codexClaudeContentBlock) { + t.Helper() + + if len(blocks) != 2 { + t.Fatalf("content block count = %d, want 2", len(blocks)) + } + expectedIDs := []string{"call_a", "call_b"} + expectedArguments := []string{`{"file_path":"a"}`, `{"file_path":"b"}`} + for index, block := range blocks { + if block.Index != int64(index) { + t.Fatalf("block %d index = %d, want %d", index, block.Index, index) + } + if block.Type != "tool_use" || block.Name != "Read" { + t.Fatalf("block %d = %#v, want Read tool_use", index, block) + } + if block.ID != expectedIDs[index] { + t.Fatalf("block %d ID = %q, want %q", index, block.ID, expectedIDs[index]) + } + if block.Arguments != expectedArguments[index] { + t.Fatalf("block %d arguments = %q, want %q", index, block.Arguments, expectedArguments[index]) + } + } +} + +func TestConvertCodexResponseToClaude_StreamSerializesInterleavedNamedFunctionCalls(t *testing.T) { + tests := []struct { + name string + chunks [][]byte + }{ + { + name: "first call finishes first", + chunks: [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_parallel","model":"gpt-5"}}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a","name":"Read"},"output_index":1}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_b","name":"Read"},"output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":\"a\"}","output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":\"b\"}","output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"a\"}","output_index":1}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},"output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"b\"}","output_index":2}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_b","name":"Read","arguments":"{\"file_path\":\"b\"}"},"output_index":2}`), + }, + }, + { + name: "second call finishes first", + chunks: [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_parallel","model":"gpt-5"}}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a","name":"Read"},"output_index":1}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_b","name":"Read"},"output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":\"b\"}","output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"b\"}","output_index":2}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_b","name":"Read","arguments":"{\"file_path\":\"b\"}"},"output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":\"a\"}","output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"a\"}","output_index":1}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},"output_index":1}`), + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + blocks := assertCodexClaudeContentBlockLifecycle(t, translateCodexClaudeChunks(t, test.chunks)) + assertParallelCodexClaudeToolCalls(t, blocks) + }) + } +} + +func TestConvertCodexResponseToClaude_StreamDefersOtherContentUntilFunctionCallsClose(t *testing.T) { + tests := []struct { + name string + functionCall []byte + firstBlock string + secondBlock string + }{ + { + name: "named active call", + functionCall: []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a","name":"Read"},"output_index":0}`), + firstBlock: "tool_use", + secondBlock: "text", + }, + { + name: "unnamed pending call", + functionCall: []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a"},"output_index":0}`), + firstBlock: "text", + secondBlock: "tool_use", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + chunks := [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_mixed","model":"gpt-5"}}`), + test.functionCall, + []byte(`data: {"type":"response.output_item.added","item":{"type":"message","status":"in_progress"},"output_index":1}`), + []byte(`data: {"type":"response.content_part.added","part":{"type":"output_text"},"content_index":0,"output_index":1}`), + []byte(`data: {"type":"response.output_text.delta","delta":"done","output_index":1}`), + []byte(`data: {"type":"response.content_part.done","part":{"type":"output_text"},"content_index":0,"output_index":1}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"message","status":"completed"},"output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"a\"}","output_index":0}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},"output_index":0}`), + []byte(`data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1}}}`), + } + + blocks := assertCodexClaudeContentBlockLifecycle(t, translateCodexClaudeChunks(t, chunks)) + if len(blocks) != 2 { + t.Fatalf("content block count = %d, want 2", len(blocks)) + } + if blocks[0].Index != 0 || blocks[0].Type != test.firstBlock { + t.Fatalf("unexpected first block: %#v", blocks[0]) + } + if blocks[1].Index != 1 || blocks[1].Type != test.secondBlock { + t.Fatalf("unexpected second block: %#v", blocks[1]) + } + for _, block := range blocks { + switch block.Type { + case "tool_use": + if block.Arguments != `{"file_path":"a"}` { + t.Fatalf("unexpected tool block: %#v", block) + } + case "text": + if block.Text != "done" { + t.Fatalf("unexpected text block: %#v", block) + } + } + } + }) + } +} + +func TestConvertCodexResponseToClaude_StreamDeferredTextClosesBeforeThinkingStarts(t *testing.T) { + chunks := [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_mixed","model":"gpt-5"}}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a","name":"Read"},"output_index":0}`), + []byte(`data: {"type":"response.content_part.added","part":{"type":"output_text"},"content_index":0,"output_index":1}`), + []byte(`data: {"type":"response.output_text.delta","delta":"answer","output_index":1}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"reasoning","encrypted_content":"enc_initial"},"output_index":2}`), + []byte(`data: {"type":"response.reasoning_summary_part.added","output_index":2}`), + []byte(`data: {"type":"response.reasoning_summary_text.delta","delta":"thought","output_index":2}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"reasoning","encrypted_content":"enc_final"},"output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"a\"}","output_index":0}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},"output_index":0}`), + []byte(`data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1}}}`), + } + + blocks := assertCodexClaudeContentBlockLifecycle(t, translateCodexClaudeChunks(t, chunks)) + if len(blocks) != 3 { + t.Fatalf("content block count = %d, want 3", len(blocks)) + } + if blocks[0].Index != 0 || blocks[0].Type != "tool_use" || blocks[0].Arguments != `{"file_path":"a"}` { + t.Fatalf("unexpected tool block: %#v", blocks[0]) + } + if blocks[1].Index != 1 || blocks[1].Type != "text" || blocks[1].Text != "answer" { + t.Fatalf("unexpected text block: %#v", blocks[1]) + } + if blocks[2].Index != 2 || blocks[2].Type != "thinking" { + t.Fatalf("unexpected thinking block: %#v", blocks[2]) + } +} + +func TestConvertCodexResponseToClaude_StreamTerminalMatchesFunctionCallsByOutputIndex(t *testing.T) { + chunks := [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_parallel","model":"gpt-5"}}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","name":"Read"},"output_index":0}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","name":"Read"},"output_index":1}`), + []byte(`data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1},"output":[{"type":"function_call","name":"Read","arguments":"{\"file_path\":\"a\"}"},{"type":"function_call","name":"Read","arguments":"{\"file_path\":\"b\"}"}]}}`), + } + + blocks := assertCodexClaudeContentBlockLifecycle(t, translateCodexClaudeChunks(t, chunks)) + if len(blocks) != 2 { + t.Fatalf("content block count = %d, want 2", len(blocks)) + } + if blocks[0].Index != 0 || blocks[0].Arguments != `{"file_path":"a"}` { + t.Fatalf("unexpected first function call: %#v", blocks[0]) + } + if blocks[1].Index != 1 || blocks[1].Arguments != `{"file_path":"b"}` { + t.Fatalf("unexpected second function call: %#v", blocks[1]) + } +} + +func TestConvertCodexResponseToClaude_StreamTerminalHydratesInterleavedFunctionCalls(t *testing.T) { + for _, terminalType := range []string{"response.completed", "response.incomplete"} { + t.Run(terminalType, func(t *testing.T) { + terminal := `data: {"type":"` + terminalType + `","response":{"usage":{"input_tokens":1,"output_tokens":1},"output":[{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},{"type":"function_call","call_id":"call_b","name":"Read","arguments":"{\"file_path\":\"b\"}"}]}}` + chunks := [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_parallel","model":"gpt-5"}}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a","name":"Read"},"output_index":0}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_b","name":"Read"},"output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":","output_index":0}`), + []byte(terminal), + } + + blocks := assertCodexClaudeContentBlockLifecycle(t, translateCodexClaudeChunks(t, chunks)) + assertParallelCodexClaudeToolCalls(t, blocks) + }) + } +} diff --git a/internal/translator/codex/claude/codex_claude_response.go b/internal/translator/codex/claude/codex_claude_response.go index 69d495832..45efd8a6c 100644 --- a/internal/translator/codex/claude/codex_claude_response.go +++ b/internal/translator/codex/claude/codex_claude_response.go @@ -27,29 +27,34 @@ const codexThinkingSummaryPartSeparator = "\n\n" // ConvertCodexResponseToClaudeParams holds parameters for response conversion. type ConvertCodexResponseToClaudeParams struct { - HasEmittedToolUse bool - BlockIndex int - HasReceivedArgumentsDelta bool - FunctionCallBlockOpen bool - FunctionCallBlockCallID string - FunctionCallBlockIndex int - HasTextDelta bool - TextBlockOpen bool - ThinkingBlockOpen bool - ThinkingSignature string - ThinkingSummarySeen bool - WebSearchToolUseIDs map[string]struct{} - WebSearchToolResultIDs map[string]struct{} - LastWebSearchToolUseID string - PendingFunctionCalls map[string]*pendingCodexFunctionCall - LastPendingFunctionCallKey string + HasEmittedToolUse bool + BlockIndex int + HasTextDelta bool + TextBlockOpen bool + ThinkingBlockOpen bool + ThinkingSignature string + ThinkingSummarySeen bool + WebSearchToolUseIDs map[string]struct{} + WebSearchToolResultIDs map[string]struct{} + LastWebSearchToolUseID string + FunctionCalls map[string]*codexFunctionCallStream + FunctionCallQueue []*codexFunctionCallStream + ActiveFunctionCall *codexFunctionCallStream + LastFunctionCall *codexFunctionCallStream + DeferredStreamEvents [][]byte } -type pendingCodexFunctionCall struct { +type codexFunctionCallStream struct { CallID string + Name string + BlockIndex int Arguments string + EmittedArgumentsLength int HasReceivedArgumentsDelta bool - StartEmitted bool + EmitInitialEmptyDelta bool + Started bool + Done bool + Closed bool } // ConvertCodexResponseToClaude performs sophisticated streaming response format conversion. @@ -78,6 +83,7 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa if !bytes.HasPrefix(rawJSON, dataTag) { return [][]byte{} } + streamEventRawJSON := bytes.Clone(rawJSON) rawJSON = bytes.TrimSpace(rawJSON[5:]) output := make([]byte, 0, 512) @@ -86,6 +92,10 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa typeResult := rootResult.Get("type") typeStr := typeResult.String() + if params.ActiveFunctionCall != nil && shouldDeferCodexStreamEvent(typeStr, rootResult) { + params.DeferredStreamEvents = append(params.DeferredStreamEvents, streamEventRawJSON) + return [][]byte{} + } var template []byte switch typeStr { @@ -98,6 +108,7 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa output = translatorcommon.AppendSSEEventBytes(output, "message_start", template, 2) case "response.reasoning_summary_part.added": + output = append(output, stopCodexTextBlock(params)...) // Codex splits a single reasoning item into several summary parts, but only // output_item.done carries that item's final encrypted_content. Keep one // thinking block open for the whole item and separate the parts with a blank @@ -109,6 +120,7 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa } params.ThinkingSummarySeen = true case "response.reasoning_summary_text.delta": + output = append(output, stopCodexTextBlock(params)...) output = append(output, startCodexThinkingBlock(params)...) output = append(output, appendCodexThinkingDelta(params, rootResult.Get("delta").String())...) case "response.reasoning_summary_part.done": @@ -137,9 +149,12 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa case "response.completed", "response.incomplete": template = []byte(`{"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"input_tokens":0,"output_tokens":0}}`) responseData := rootResult.Get("response") - output = hydrateOpenCodexFunctionCallFromTerminal(output, params, responseData) - output = append(output, finalizeCodexOpenContentBlocks(params)...) - output = appendPendingCodexFunctionCallsFromTerminal(output, params, originalRequestRawJSON, responseData) + output = append(output, finalizeCodexThinkingBlock(params)...) + output = append(output, stopCodexTextBlock(params)...) + output = appendCodexFunctionCallsFromTerminal(output, params, originalRequestRawJSON, responseData) + output = appendDeferredCodexStreamEvents(output, originalRequestRawJSON, param) + output = append(output, finalizeCodexThinkingBlock(params)...) + output = append(output, stopCodexTextBlock(params)...) template, _ = sjson.SetBytes(template, "delta.stop_reason", mapCodexStopReasonToClaude(codexStopReason(responseData), params.HasEmittedToolUse)) template = setClaudeStopSequence(template, "delta.stop_sequence", responseData) inputTokens, outputTokens, cachedTokens := extractResponsesUsage(responseData.Get("usage")) @@ -158,26 +173,15 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa case "function_call": output = append(output, finalizeCodexThinkingBlock(params)...) output = append(output, stopCodexTextBlock(params)...) - params.HasReceivedArgumentsDelta = false - callID := codexFunctionCallID(itemResult) - name := itemResult.Get("name").String() - if name == "" { - recordPendingCodexFunctionCall(params, rootResult, itemResult) - break + call := recordCodexFunctionCall(params, rootResult, itemResult) + updateCodexFunctionCallIdentity(params, call, rootResult, itemResult) + if call.Name != "" { + call.EmitInitialEmptyDelta = true } - - if pending, pendingKeys := pendingCodexFunctionCallForDone(params, rootResult, itemResult); pending != nil { - deletePendingCodexFunctionCallAliases(params, pendingKeys) - } - blockIndex := params.BlockIndex - output = appendCodexFunctionCallStart(output, originalRequestRawJSON, callID, name, blockIndex) - params.HasEmittedToolUse = true - output = appendCodexFunctionCallArgumentDelta(output, "", blockIndex) - params.FunctionCallBlockOpen = true - params.FunctionCallBlockCallID = callID - params.FunctionCallBlockIndex = blockIndex + output = appendCodexFunctionCallQueue(output, params, originalRequestRawJSON) case "reasoning": + output = append(output, stopCodexTextBlock(params)...) // A previous reasoning item that never reported output_item.done must not // leak its still-open block into this one. output = append(output, finalizeCodexThinkingBlock(params)...) @@ -226,41 +230,18 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa output = append(output, stopCodexTextBlock(params)...) params.HasTextDelta = true case "function_call": - if pending, pendingKeys := pendingCodexFunctionCallForDone(params, rootResult, itemResult); pending != nil && !pending.StartEmitted { - name := itemResult.Get("name").String() - if name == "" { - return [][]byte{output} - } - callID := pending.CallID - if callID == "" { - callID = codexFunctionCallID(itemResult) - } - blockIndex := params.BlockIndex - output = appendCodexFunctionCallStart(output, originalRequestRawJSON, callID, name, blockIndex) - params.HasEmittedToolUse = true - pending.StartEmitted = true - - args := pending.Arguments - if args == "" { - args = itemResult.Get("arguments").String() - } - if args != "" { - output = appendCodexFunctionCallArgumentDelta(output, args, blockIndex) - } - output = appendCodexFunctionCallStop(output, blockIndex) - params.BlockIndex++ - - deletePendingCodexFunctionCallAliases(params, pendingKeys) - } else if params.FunctionCallBlockOpen { - if !params.HasReceivedArgumentsDelta { - if args := itemResult.Get("arguments").String(); args != "" { - output = appendCodexFunctionCallArgumentDelta(output, args, params.FunctionCallBlockIndex) - params.HasReceivedArgumentsDelta = true - } - } - output = appendCodexOpenFunctionCallStop(output, params) + output = append(output, finalizeCodexThinkingBlock(params)...) + output = append(output, stopCodexTextBlock(params)...) + call := codexFunctionCallForEvent(params, rootResult, itemResult) + if call == nil { + call = recordCodexFunctionCall(params, rootResult, itemResult) } + updateCodexFunctionCallIdentity(params, call, rootResult, itemResult) + updateCodexFunctionCallArguments(call, itemResult.Get("arguments").String(), false) + call.Done = true + output = appendCodexFunctionCallQueue(output, params, originalRequestRawJSON) case "reasoning": + output = append(output, stopCodexTextBlock(params)...) if signature := itemResult.Get("encrypted_content").String(); signature != "" { params.ThinkingSignature = signature } @@ -275,36 +256,58 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa output = appendCodexWebSearchToolResult(output, params, rootResult, itemResult) } case "response.function_call_arguments.delta": - delta := rootResult.Get("delta").String() - key := codexArgumentsFunctionCallKey(params, rootResult) - if pending, _ := pendingCodexFunctionCallForKey(params, key); pending != nil && !pending.StartEmitted { - pending.HasReceivedArgumentsDelta = true - pending.Arguments += delta - break + call := codexFunctionCallForEvent(params, rootResult, gjson.Result{}) + if call == nil { + call = recordCodexFunctionCall(params, rootResult, gjson.Result{}) } - - params.HasReceivedArgumentsDelta = true - output = appendCodexFunctionCallArgumentDelta(output, delta, params.BlockIndex) + updateCodexFunctionCallArguments(call, rootResult.Get("delta").String(), true) + output = appendCodexFunctionCallBufferedArguments(output, params, call) case "response.function_call_arguments.done": - key := codexArgumentsFunctionCallKey(params, rootResult) - if pending, _ := pendingCodexFunctionCallForKey(params, key); pending != nil && !pending.StartEmitted { - if !pending.HasReceivedArgumentsDelta { - pending.Arguments = rootResult.Get("arguments").String() - } - break - } - - if !params.HasReceivedArgumentsDelta { - if args := rootResult.Get("arguments").String(); args != "" { - output = appendCodexFunctionCallArgumentDelta(output, args, params.BlockIndex) - params.HasReceivedArgumentsDelta = true - } + call := codexFunctionCallForEvent(params, rootResult, gjson.Result{}) + if call == nil { + call = recordCodexFunctionCall(params, rootResult, gjson.Result{}) } + updateCodexFunctionCallArguments(call, rootResult.Get("arguments").String(), false) + output = appendCodexFunctionCallBufferedArguments(output, params, call) } + if len(params.FunctionCallQueue) == 0 { + output = appendDeferredCodexStreamEvents(output, originalRequestRawJSON, param) + } return [][]byte{output} } +func shouldDeferCodexStreamEvent(typeStr string, rootResult gjson.Result) bool { + switch typeStr { + case "error", "response.completed", "response.incomplete", "response.function_call_arguments.delta", "response.function_call_arguments.done": + return false + case "response.output_item.added", "response.output_item.done": + return rootResult.Get("item.type").String() != "function_call" + default: + return true + } +} + +func appendDeferredCodexStreamEvents(output []byte, originalRequestRawJSON []byte, param *any) []byte { + if param == nil || *param == nil { + return output + } + params := (*param).(*ConvertCodexResponseToClaudeParams) + if len(params.DeferredStreamEvents) == 0 { + return output + } + + events := params.DeferredStreamEvents + params.DeferredStreamEvents = nil + for _, event := range events { + translated := ConvertCodexResponseToClaude(context.Background(), "", originalRequestRawJSON, nil, event, param) + for _, chunk := range translated { + output = append(output, chunk...) + } + } + return output +} + func codexStreamErrorToClaudeError(rootResult gjson.Result) []byte { errorResult := rootResult.Get("error") errType := strings.TrimSpace(errorResult.Get("type").String()) @@ -515,78 +518,28 @@ func setClaudeStopSequence(out []byte, path string, responseData gjson.Result) [ return out } -func codexFunctionCallKey(rootResult, itemResult gjson.Result) string { - if outputIndex := rootResult.Get("output_index"); outputIndex.Exists() { - return "output:" + outputIndex.Raw - } - if callID := codexFunctionCallID(itemResult); callID != "" { - return "call:" + callID - } - return "last" -} - func codexFunctionCallID(itemResult gjson.Result) string { return itemResult.Get("call_id").String() } -func codexFunctionCallIDKey(callID string) string { - if callID == "" { - return "" - } - return "call:" + callID -} - -func codexArgumentsFunctionCallKey(params *ConvertCodexResponseToClaudeParams, rootResult gjson.Result) string { +func codexFunctionCallKeys(rootResult, itemResult gjson.Result) []string { + keys := make([]string, 0, 5) if outputIndex := rootResult.Get("output_index"); outputIndex.Exists() { - return "output:" + outputIndex.Raw + keys = appendUniqueCodexFunctionCallKey(keys, "output:"+outputIndex.Raw) } - return params.LastPendingFunctionCallKey -} - -func recordPendingCodexFunctionCall(params *ConvertCodexResponseToClaudeParams, rootResult, itemResult gjson.Result) { - if params.PendingFunctionCalls == nil { - params.PendingFunctionCalls = map[string]*pendingCodexFunctionCall{} + if callID := codexFunctionCallID(itemResult); callID != "" { + keys = appendUniqueCodexFunctionCallKey(keys, "call:"+callID) } - - pending := &pendingCodexFunctionCall{CallID: codexFunctionCallID(itemResult)} - key := codexFunctionCallKey(rootResult, itemResult) - params.PendingFunctionCalls[key] = pending - if callIDKey := codexFunctionCallIDKey(pending.CallID); callIDKey != "" { - params.PendingFunctionCalls[callIDKey] = pending + if callID := rootResult.Get("call_id").String(); callID != "" { + keys = appendUniqueCodexFunctionCallKey(keys, "call:"+callID) } - params.LastPendingFunctionCallKey = key -} - -func pendingCodexFunctionCallForKey(params *ConvertCodexResponseToClaudeParams, key string) (*pendingCodexFunctionCall, string) { - if params == nil || params.PendingFunctionCalls == nil || key == "" { - return nil, "" + if itemID := itemResult.Get("id").String(); itemID != "" { + keys = appendUniqueCodexFunctionCallKey(keys, "item:"+itemID) } - pending, ok := params.PendingFunctionCalls[key] - if !ok { - return nil, "" + if itemID := rootResult.Get("item_id").String(); itemID != "" { + keys = appendUniqueCodexFunctionCallKey(keys, "item:"+itemID) } - return pending, key -} - -func pendingCodexFunctionCallForDone(params *ConvertCodexResponseToClaudeParams, rootResult, itemResult gjson.Result) (*pendingCodexFunctionCall, []string) { - if params == nil || params.PendingFunctionCalls == nil { - return nil, nil - } - - keys := []string{codexFunctionCallKey(rootResult, itemResult)} - callID := codexFunctionCallID(itemResult) - if callID != "" { - keys = appendUniqueCodexFunctionCallKey(keys, codexFunctionCallIDKey(callID)) - } else if !rootResult.Get("output_index").Exists() && params.LastPendingFunctionCallKey != "" { - keys = appendUniqueCodexFunctionCallKey(keys, params.LastPendingFunctionCallKey) - } - - for _, key := range keys { - if pending, ok := params.PendingFunctionCalls[key]; ok { - return pending, keysForPendingCodexFunctionCall(params, pending) - } - } - return nil, nil + return keys } func appendUniqueCodexFunctionCallKey(keys []string, key string) []string { @@ -601,29 +554,81 @@ func appendUniqueCodexFunctionCallKey(keys []string, key string) []string { return append(keys, key) } -func keysForPendingCodexFunctionCall(params *ConvertCodexResponseToClaudeParams, pending *pendingCodexFunctionCall) []string { - if params == nil || pending == nil || params.PendingFunctionCalls == nil { +func codexFunctionCallForKeys(params *ConvertCodexResponseToClaudeParams, keys []string) *codexFunctionCallStream { + if params == nil || params.FunctionCalls == nil { return nil } - - keys := make([]string, 0, 2) - for key, candidate := range params.PendingFunctionCalls { - if candidate == pending { - keys = append(keys, key) + for _, key := range keys { + if call := params.FunctionCalls[key]; call != nil { + return call } } - return keys + return nil } -func deletePendingCodexFunctionCallAliases(params *ConvertCodexResponseToClaudeParams, keys []string) { - if params == nil || params.PendingFunctionCalls == nil { +func codexFunctionCallForEvent(params *ConvertCodexResponseToClaudeParams, rootResult, itemResult gjson.Result) *codexFunctionCallStream { + keys := codexFunctionCallKeys(rootResult, itemResult) + if len(keys) > 0 { + return codexFunctionCallForKeys(params, keys) + } + if params == nil { + return nil + } + return params.LastFunctionCall +} + +func recordCodexFunctionCall(params *ConvertCodexResponseToClaudeParams, rootResult, itemResult gjson.Result) *codexFunctionCallStream { + keys := codexFunctionCallKeys(rootResult, itemResult) + call := codexFunctionCallForKeys(params, keys) + if call == nil { + call = &codexFunctionCallStream{BlockIndex: -1} + params.FunctionCallQueue = append(params.FunctionCallQueue, call) + } + addCodexFunctionCallAliases(params, call, keys) + params.LastFunctionCall = call + return call +} + +func addCodexFunctionCallAliases(params *ConvertCodexResponseToClaudeParams, call *codexFunctionCallStream, keys []string) { + if params == nil || call == nil { return } + if params.FunctionCalls == nil { + params.FunctionCalls = map[string]*codexFunctionCallStream{} + } for _, key := range keys { - delete(params.PendingFunctionCalls, key) - if params.LastPendingFunctionCallKey == key { - params.LastPendingFunctionCallKey = "" - } + params.FunctionCalls[key] = call + } +} + +func updateCodexFunctionCallIdentity(params *ConvertCodexResponseToClaudeParams, call *codexFunctionCallStream, rootResult, itemResult gjson.Result) { + if call == nil { + return + } + if callID := codexFunctionCallID(itemResult); callID != "" { + call.CallID = callID + } + if name := itemResult.Get("name").String(); name != "" { + call.Name = name + } + addCodexFunctionCallAliases(params, call, codexFunctionCallKeys(rootResult, itemResult)) +} + +func updateCodexFunctionCallArguments(call *codexFunctionCallStream, arguments string, delta bool) { + if call == nil || arguments == "" { + return + } + if delta { + call.Arguments += arguments + call.HasReceivedArgumentsDelta = true + return + } + if !call.HasReceivedArgumentsDelta { + call.Arguments = arguments + return + } + if strings.HasPrefix(arguments, call.Arguments) { + call.Arguments = arguments } } @@ -648,42 +653,78 @@ func appendCodexFunctionCallStop(output []byte, blockIndex int) []byte { return translatorcommon.AppendSSEEventBytes(output, "content_block_stop", template, 2) } -func appendCodexOpenFunctionCallStop(output []byte, params *ConvertCodexResponseToClaudeParams) []byte { - if params == nil || !params.FunctionCallBlockOpen { +func appendCodexFunctionCallBufferedArguments(output []byte, params *ConvertCodexResponseToClaudeParams, call *codexFunctionCallStream) []byte { + if params == nil || call == nil || params.ActiveFunctionCall != call || !call.Started || call.Closed { + return output + } + if call.EmittedArgumentsLength >= len(call.Arguments) { return output } - blockIndex := params.FunctionCallBlockIndex - output = appendCodexFunctionCallStop(output, blockIndex) - if params.BlockIndex <= blockIndex { - params.BlockIndex = blockIndex + 1 - } - params.FunctionCallBlockOpen = false - params.FunctionCallBlockCallID = "" - params.FunctionCallBlockIndex = 0 + output = appendCodexFunctionCallArgumentDelta(output, call.Arguments[call.EmittedArgumentsLength:], call.BlockIndex) + call.EmittedArgumentsLength = len(call.Arguments) return output } -func hydrateOpenCodexFunctionCallFromTerminal(output []byte, params *ConvertCodexResponseToClaudeParams, responseData gjson.Result) []byte { - if params == nil || !params.FunctionCallBlockOpen || params.HasReceivedArgumentsDelta { +func appendCodexFunctionCallQueue(output []byte, params *ConvertCodexResponseToClaudeParams, originalRequestRawJSON []byte) []byte { + if params == nil { return output } - responseData.Get("output").ForEach(func(_, item gjson.Result) bool { - if item.Get("type").String() != "function_call" || codexFunctionCallID(item) != params.FunctionCallBlockCallID { - return true + for { + if active := params.ActiveFunctionCall; active != nil { + output = appendCodexFunctionCallBufferedArguments(output, params, active) + if !active.Done { + return output + } + output = appendCodexFunctionCallStop(output, active.BlockIndex) + if params.BlockIndex <= active.BlockIndex { + params.BlockIndex = active.BlockIndex + 1 + } + active.Closed = true + params.ActiveFunctionCall = nil + removeCodexFunctionCallFromQueue(params, active) } - if args := item.Get("arguments").String(); args != "" { - output = appendCodexFunctionCallArgumentDelta(output, args, params.FunctionCallBlockIndex) - params.HasReceivedArgumentsDelta = true + + for len(params.FunctionCallQueue) > 0 && params.FunctionCallQueue[0].Closed { + params.FunctionCallQueue = params.FunctionCallQueue[1:] } - return false - }) - return output + if len(params.FunctionCallQueue) == 0 { + return output + } + + call := params.FunctionCallQueue[0] + if call.Name == "" { + return output + } + + call.BlockIndex = params.BlockIndex + output = appendCodexFunctionCallStart(output, originalRequestRawJSON, call.CallID, call.Name, call.BlockIndex) + if call.EmitInitialEmptyDelta { + output = appendCodexFunctionCallArgumentDelta(output, "", call.BlockIndex) + } + call.Started = true + params.ActiveFunctionCall = call + params.HasEmittedToolUse = true + output = appendCodexFunctionCallBufferedArguments(output, params, call) + } } -func appendPendingCodexFunctionCallsFromTerminal(output []byte, params *ConvertCodexResponseToClaudeParams, originalRequestRawJSON []byte, responseData gjson.Result) []byte { - if params == nil || len(params.PendingFunctionCalls) == 0 { +func removeCodexFunctionCallFromQueue(params *ConvertCodexResponseToClaudeParams, call *codexFunctionCallStream) { + if params == nil || call == nil { + return + } + for index, queued := range params.FunctionCallQueue { + if queued != call { + continue + } + params.FunctionCallQueue = append(params.FunctionCallQueue[:index], params.FunctionCallQueue[index+1:]...) + return + } +} + +func appendCodexFunctionCallsFromTerminal(output []byte, params *ConvertCodexResponseToClaudeParams, originalRequestRawJSON []byte, responseData gjson.Result) []byte { + if params == nil { return output } @@ -692,88 +733,52 @@ func appendPendingCodexFunctionCallsFromTerminal(output []byte, params *ConvertC return true } - pending, pendingKeys := pendingCodexFunctionCallForTerminalItem(params, index, item) - if pending == nil { - return true + keys := codexFunctionCallKeys(gjson.Result{}, item) + if itemOutputIndex := item.Get("output_index"); itemOutputIndex.Exists() { + keys = appendUniqueCodexFunctionCallKey(keys, "output:"+itemOutputIndex.Raw) } - if pending.StartEmitted { - deletePendingCodexFunctionCallAliases(params, pendingKeys) - return true + if index.Exists() { + keys = appendUniqueCodexFunctionCallKey(keys, "output:"+index.String()) } - - name := item.Get("name").String() - if name == "" { - deletePendingCodexFunctionCallAliases(params, pendingKeys) - return true + call := codexFunctionCallForKeys(params, keys) + if call == nil { + call = &codexFunctionCallStream{BlockIndex: -1} + params.FunctionCallQueue = append(params.FunctionCallQueue, call) } - callID := pending.CallID - if callID == "" { - callID = codexFunctionCallID(item) - } - - blockIndex := params.BlockIndex - output = appendCodexFunctionCallStart(output, originalRequestRawJSON, callID, name, blockIndex) - params.HasEmittedToolUse = true - pending.StartEmitted = true - - args := item.Get("arguments").String() - if args == "" { - args = pending.Arguments - } - if args != "" { - output = appendCodexFunctionCallArgumentDelta(output, args, blockIndex) - } - output = appendCodexFunctionCallStop(output, blockIndex) - params.BlockIndex++ - - deletePendingCodexFunctionCallAliases(params, pendingKeys) + addCodexFunctionCallAliases(params, call, keys) + updateCodexFunctionCallIdentity(params, call, gjson.Result{}, item) + updateCodexFunctionCallArguments(call, item.Get("arguments").String(), false) + call.Done = true return true }) - clearPendingCodexFunctionCalls(params) + queuedCalls := params.FunctionCallQueue[:0] + for _, call := range params.FunctionCallQueue { + if call.Closed { + continue + } + if call.Name == "" { + call.Closed = true + continue + } + call.Done = true + queuedCalls = append(queuedCalls, call) + } + params.FunctionCallQueue = queuedCalls + output = appendCodexFunctionCallQueue(output, params, originalRequestRawJSON) + + clearCodexFunctionCalls(params) return output } -func pendingCodexFunctionCallForTerminalItem(params *ConvertCodexResponseToClaudeParams, outputIndex, item gjson.Result) (*pendingCodexFunctionCall, []string) { - if params == nil || params.PendingFunctionCalls == nil { - return nil, nil - } - - keys := make([]string, 0, 3) - if callID := codexFunctionCallID(item); callID != "" { - keys = appendUniqueCodexFunctionCallKey(keys, codexFunctionCallIDKey(callID)) - } - if itemOutputIndex := item.Get("output_index"); itemOutputIndex.Exists() { - keys = appendUniqueCodexFunctionCallKey(keys, "output:"+itemOutputIndex.Raw) - } - if outputIndex.Exists() { - keys = appendUniqueCodexFunctionCallKey(keys, "output:"+outputIndex.Raw) - } - - for _, key := range keys { - if pending, ok := params.PendingFunctionCalls[key]; ok { - return pending, keysForPendingCodexFunctionCall(params, pending) - } - } - return nil, nil -} - -func clearPendingCodexFunctionCalls(params *ConvertCodexResponseToClaudeParams) { - if params == nil || params.PendingFunctionCalls == nil { +func clearCodexFunctionCalls(params *ConvertCodexResponseToClaudeParams) { + if params == nil { return } - for key := range params.PendingFunctionCalls { - delete(params.PendingFunctionCalls, key) - } - params.LastPendingFunctionCallKey = "" -} - -func finalizeCodexOpenContentBlocks(params *ConvertCodexResponseToClaudeParams) []byte { - output := make([]byte, 0, 256) - output = append(output, finalizeCodexThinkingBlock(params)...) - output = append(output, stopCodexTextBlock(params)...) - output = appendCodexOpenFunctionCallStop(output, params) - return output + clear(params.FunctionCalls) + params.FunctionCallQueue = nil + params.ActiveFunctionCall = nil + params.LastFunctionCall = nil } func resolveCodexClaudeToolUseName(originalRequestRawJSON []byte, name string) string { diff --git a/internal/translator/codex/claude/codex_claude_response_test.go b/internal/translator/codex/claude/codex_claude_response_test.go index bc59cec34..3ed49a4db 100644 --- a/internal/translator/codex/claude/codex_claude_response_test.go +++ b/internal/translator/codex/claude/codex_claude_response_test.go @@ -879,7 +879,7 @@ func TestConvertCodexResponseToClaude_StreamUnresolvedPendingFunctionCallDoesNot t.Fatalf("stop_reason = %q, want end_turn. Outputs=%q", gotReason, outputs) } params, ok := param.(*ConvertCodexResponseToClaudeParams) - if !ok || len(params.PendingFunctionCalls) != 0 || params.LastPendingFunctionCallKey != "" { + if !ok || len(params.FunctionCalls) != 0 || len(params.FunctionCallQueue) != 0 || params.LastFunctionCall != nil { t.Fatalf("pending function calls were not cleared: %#v", param) } } diff --git a/internal/translator/codex/claude/codex_claude_response_web_search.go b/internal/translator/codex/claude/codex_claude_response_web_search.go index 1f9c59a7c..c5c8f866f 100644 --- a/internal/translator/codex/claude/codex_claude_response_web_search.go +++ b/internal/translator/codex/claude/codex_claude_response_web_search.go @@ -28,6 +28,7 @@ func appendCodexWebSearchServerToolUse(output []byte, params *ConvertCodexRespon } if !alreadyStarted { + output = append(output, stopCodexTextBlock(params)...) output = append(output, finalizeCodexThinkingBlock(params)...) template := []byte(`{"type":"content_block_start","index":0,"content_block":{"type":"server_tool_use","id":"","name":"web_search","input":{}}}`) template, _ = sjson.SetBytes(template, "index", params.BlockIndex) diff --git a/test/codex_claude_parallel_function_calls_test.go b/test/codex_claude_parallel_function_calls_test.go new file mode 100644 index 000000000..782551905 --- /dev/null +++ b/test/codex_claude_parallel_function_calls_test.go @@ -0,0 +1,125 @@ +package test + +import ( + "context" + "strings" + "testing" + + _ "github.com/router-for-me/CLIProxyAPI/v7/internal/translator" + sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" + "github.com/tidwall/gjson" +) + +func TestCodexToClaudeParallelFunctionCallsHaveValidLifecycle(t *testing.T) { + chunks := [][]byte{ + []byte(`data: {"type":"response.created","response":{"id":"resp_parallel","model":"gpt-5"}}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_a","name":"Read"},"output_index":1}`), + []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_b","name":"Read"},"output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":\"a\"}","output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"a\"}","output_index":1}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},"output_index":1}`), + []byte(`data: {"type":"response.function_call_arguments.delta","delta":"{\"file_path\":\"b\"}","output_index":2}`), + []byte(`data: {"type":"response.function_call_arguments.done","arguments":"{\"file_path\":\"b\"}","output_index":2}`), + []byte(`data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_b","name":"Read","arguments":"{\"file_path\":\"b\"}"},"output_index":2}`), + []byte(`data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1},"output":[{"type":"function_call","call_id":"call_a","name":"Read","arguments":"{\"file_path\":\"a\"}"},{"type":"function_call","call_id":"call_b","name":"Read","arguments":"{\"file_path\":\"b\"}"}]}}`), + } + + originalRequest := []byte(`{"stream":true,"tools":[{"name":"Read"}]}`) + var state any + open := make(map[int64]struct{}) + started := make(map[int64]struct{}) + toolIDs := make(map[int64]string) + arguments := make(map[int64]string) + var startIndices []int64 + var stopIndices []int64 + messageState := 0 + + for _, chunk := range chunks { + outputs := sdktranslator.TranslateStream( + context.Background(), + sdktranslator.FormatCodex, + sdktranslator.FormatClaude, + "gpt-5", + originalRequest, + nil, + chunk, + &state, + ) + for _, output := range outputs { + for _, line := range strings.Split(string(output), "\n") { + if !strings.HasPrefix(line, "data: ") { + continue + } + event := gjson.Parse(strings.TrimPrefix(line, "data: ")) + if messageState == 2 { + t.Fatalf("event emitted after message_stop: %s", event.Raw) + } + index := event.Get("index").Int() + switch event.Get("type").String() { + case "content_block_start": + if messageState != 0 { + t.Fatalf("content block started after message terminal events: %s", event.Raw) + } + if len(open) != 0 { + t.Fatalf("content block start emitted while another block remains open: %v", open) + } + if _, exists := started[index]; exists { + t.Fatalf("content block index %d was reused", index) + } + open[index] = struct{}{} + started[index] = struct{}{} + startIndices = append(startIndices, index) + toolIDs[index] = event.Get("content_block.id").String() + case "content_block_delta": + if _, exists := open[index]; !exists { + t.Fatalf("content block delta targets unopened index %d", index) + } + if event.Get("delta.type").String() == "input_json_delta" { + arguments[index] += event.Get("delta.partial_json").String() + } + case "content_block_stop": + if _, exists := open[index]; !exists { + t.Fatalf("content block stop targets unopened index %d", index) + } + delete(open, index) + stopIndices = append(stopIndices, index) + case "message_delta": + if len(open) != 0 { + t.Fatalf("message_delta emitted while content blocks remain open: %v", open) + } + if messageState != 0 { + t.Fatalf("duplicate or out-of-order message_delta: %s", event.Raw) + } + messageState = 1 + case "message_stop": + if len(open) != 0 { + t.Fatalf("message_stop emitted while content blocks remain open: %v", open) + } + if messageState != 1 { + t.Fatalf("message_stop emitted before message_delta: %s", event.Raw) + } + messageState = 2 + } + } + } + } + + if len(open) != 0 { + t.Fatalf("content blocks remain open: %v", open) + } + if messageState != 2 { + t.Fatalf("terminal message event state = %d, want message_delta followed by message_stop", messageState) + } + if len(startIndices) != 2 || startIndices[0] != 0 || startIndices[1] != 1 { + t.Fatalf("start indices = %v, want [0 1]", startIndices) + } + if len(stopIndices) != 2 || stopIndices[0] != 0 || stopIndices[1] != 1 { + t.Fatalf("stop indices = %v, want [0 1]", stopIndices) + } + if toolIDs[0] != "call_a" || toolIDs[1] != "call_b" { + t.Fatalf("tool IDs = %v, want call_a and call_b", toolIDs) + } + if arguments[0] != `{"file_path":"a"}` || arguments[1] != `{"file_path":"b"}` { + t.Fatalf("tool arguments = %v", arguments) + } +}