From 6271c517da6ec76905667fb5f2bcd958d5acdfd9 Mon Sep 17 00:00:00 2001 From: David Stone Date: Fri, 4 Sep 2026 21:18:36 -0600 Subject: [PATCH] fix(apicompat): preserve arguments from done events --- .../chatcompletions_responses_test.go | 102 ++++++++++++++++++ .../apicompat/responses_to_chatcompletions.go | 87 +++++++++++++-- 2 files changed, 178 insertions(+), 11 deletions(-) diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_test.go b/backend/internal/pkg/apicompat/chatcompletions_responses_test.go index a57bd73f91..1731892735 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_test.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_test.go @@ -1621,6 +1621,108 @@ func TestBufferedResponseAccumulator_ToolCalls(t *testing.T) { assert.Equal(t, `{"city":"NYC"}`, output[0].Arguments) } +func TestResponsesEventToChatChunks_FunctionArgumentsDoneWithoutDeltas(t *testing.T) { + state := NewResponsesEventToChatState() + state.Model = "gpt-5-codex" + state.SentRole = true + + chunks := ResponsesEventToChatChunks(&ResponsesStreamEvent{ + Type: "response.output_item.added", + OutputIndex: 1, + Item: &ResponsesOutput{ + Type: "function_call", + CallID: "call_lookup", + Name: "lookup_item", + Arguments: "", + }, + }, state) + require.Len(t, chunks, 1) + + chunks = ResponsesEventToChatChunks(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", + OutputIndex: 1, + CallID: "call_lookup", + Name: "lookup_item", + Arguments: `{"id":1}`, + }, state) + require.Len(t, chunks, 1) + require.Len(t, chunks[0].Choices[0].Delta.ToolCalls, 1) + assert.Equal(t, `{"id":1}`, chunks[0].Choices[0].Delta.ToolCalls[0].Function.Arguments) +} + +func TestResponsesEventToChatChunks_FunctionArgumentsDoneDoesNotDuplicateDeltas(t *testing.T) { + state := NewResponsesEventToChatState() + state.Model = "gpt-5-codex" + state.SentRole = true + + ResponsesEventToChatChunks(&ResponsesStreamEvent{ + Type: "response.output_item.added", + OutputIndex: 0, + Item: &ResponsesOutput{Type: "function_call", CallID: "call_lookup", Name: "lookup_item"}, + }, state) + chunks := ResponsesEventToChatChunks(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.delta", + OutputIndex: 0, + Delta: `{"id":1}`, + }, state) + require.Len(t, chunks, 1) + + chunks = ResponsesEventToChatChunks(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", + OutputIndex: 0, + Arguments: `{"id":1}`, + }, state) + assert.Empty(t, chunks) +} + +func TestBufferedResponseAccumulator_FunctionArgumentsDoneWithoutDeltas(t *testing.T) { + acc := NewBufferedResponseAccumulator() + acc.ProcessEvent(&ResponsesStreamEvent{ + Type: "response.output_item.added", + OutputIndex: 1, + Item: &ResponsesOutput{Type: "function_call", CallID: "call_lookup", Name: "lookup_item"}, + }) + acc.ProcessEvent(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", + OutputIndex: 1, + Arguments: `{"id":1}`, + }) + + resp := &ResponsesResponse{Output: []ResponsesOutput{{ + Type: "function_call", + CallID: "call_lookup", + Name: "lookup_item", + Arguments: "", + }}} + acc.SupplementResponseOutput(resp) + + require.Len(t, resp.Output, 1) + assert.Equal(t, `{"id":1}`, resp.Output[0].Arguments) +} + +func TestBufferedResponseAccumulator_FunctionArgumentsDoneReplacesDeltas(t *testing.T) { + acc := NewBufferedResponseAccumulator() + acc.ProcessEvent(&ResponsesStreamEvent{ + Type: "response.output_item.added", + OutputIndex: 0, + Item: &ResponsesOutput{Type: "function_call", CallID: "call_lookup", Name: "lookup_item"}, + }) + acc.ProcessEvent(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.delta", + OutputIndex: 0, + Delta: `{"id":1}`, + }) + acc.ProcessEvent(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", + OutputIndex: 0, + Arguments: `{"id":1}`, + }) + + output := acc.BuildOutput() + require.Len(t, output, 1) + assert.Equal(t, `{"id":1}`, output[0].Arguments) +} + func TestBufferedResponseAccumulator_Reasoning(t *testing.T) { acc := NewBufferedResponseAccumulator() diff --git a/backend/internal/pkg/apicompat/responses_to_chatcompletions.go b/backend/internal/pkg/apicompat/responses_to_chatcompletions.go index d288b31bea..e649a261df 100644 --- a/backend/internal/pkg/apicompat/responses_to_chatcompletions.go +++ b/backend/internal/pkg/apicompat/responses_to_chatcompletions.go @@ -126,6 +126,7 @@ type ResponsesEventToChatState struct { Finalized bool // true after finish chunk has been emitted NextToolCallIndex int // next sequential tool_call index to assign OutputIndexToToolIndex map[int]int // Responses output_index → Chat tool_calls index + OutputIndexToArguments map[int]string IncludeUsage bool Usage *ChatUsage } @@ -136,6 +137,7 @@ func NewResponsesEventToChatState() *ResponsesEventToChatState { ID: generateChatCmplID(), Created: time.Now().Unix(), OutputIndexToToolIndex: make(map[int]int), + OutputIndexToArguments: make(map[int]string), } } @@ -154,6 +156,8 @@ func ResponsesEventToChatChunks(evt *ResponsesStreamEvent, state *ResponsesEvent // 均按 OutputIndex 累加到对应工具调用。 "response.custom_tool_call_input.delta": return resToChatHandleFuncArgsDelta(evt, state) + case "response.function_call_arguments.done", "response.custom_tool_call_input.done": + return resToChatHandleFuncArgsDone(evt, state) case "response.reasoning_summary_text.delta", // 原始推理文本增量（真实 Codex 客户端消费的 reasoning_text.delta）， // 与 reasoning summary 一样映射为 reasoning_content。 @@ -277,6 +281,7 @@ func resToChatHandleFuncArgsDelta(evt *ResponsesStreamEvent, state *ResponsesEve if !ok { return nil } + state.OutputIndexToArguments[evt.OutputIndex] += evt.Delta return []ChatCompletionsChunk{makeChatDeltaChunk(state, ChatDelta{ ToolCalls: []ChatToolCall{{ @@ -288,6 +293,33 @@ func resToChatHandleFuncArgsDelta(evt *ResponsesStreamEvent, state *ResponsesEve })} } +func resToChatHandleFuncArgsDone(evt *ResponsesStreamEvent, state *ResponsesEventToChatState) []ChatCompletionsChunk { + idx, ok := state.OutputIndexToToolIndex[evt.OutputIndex] + if !ok { + return nil + } + + completed := evt.Arguments + if evt.Type == "response.custom_tool_call_input.done" { + completed = evt.Input + } + current := state.OutputIndexToArguments[evt.OutputIndex] + if completed == "" || !strings.HasPrefix(completed, current) || completed == current { + return nil + } + + remainder := completed[len(current):] + state.OutputIndexToArguments[evt.OutputIndex] = completed + return []ChatCompletionsChunk{makeChatDeltaChunk(state, ChatDelta{ + ToolCalls: []ChatToolCall{{ + Index: &idx, + Function: ChatFunctionCall{ + Arguments: remainder, + }, + }}, + })} +} + func resToChatHandleReasoningDelta(evt *ResponsesStreamEvent, state *ResponsesEventToChatState) []ChatCompletionsChunk { if evt.Delta == "" { return nil @@ -453,9 +485,10 @@ func generateChatCmplID() string { // --------------------------------------------------------------------------- type bufferedFuncCall struct { - CallID string - Name string - Args strings.Builder + OutputIndex int + CallID string + Name string + Args strings.Builder } // BufferedResponseAccumulator collects content from Responses SSE delta events @@ -489,8 +522,9 @@ func (a *BufferedResponseAccumulator) ProcessEvent(event *ResponsesStreamEvent) idx := len(a.funcCalls) a.outputIndexToFuncIdx[event.OutputIndex] = idx a.funcCalls = append(a.funcCalls, bufferedFuncCall{ - CallID: event.Item.CallID, - Name: event.Item.Name, + OutputIndex: event.OutputIndex, + CallID: event.Item.CallID, + Name: event.Item.Name, }) } case "response.function_call_arguments.delta", "response.custom_tool_call_input.delta": @@ -499,6 +533,17 @@ func (a *BufferedResponseAccumulator) ProcessEvent(event *ResponsesStreamEvent) _, _ = a.funcCalls[idx].Args.WriteString(event.Delta) } } + case "response.function_call_arguments.done", "response.custom_tool_call_input.done": + completed := event.Arguments + if event.Type == "response.custom_tool_call_input.done" { + completed = event.Input + } + if completed != "" { + if idx, ok := a.outputIndexToFuncIdx[event.OutputIndex]; ok { + a.funcCalls[idx].Args.Reset() + _, _ = a.funcCalls[idx].Args.WriteString(completed) + } + } case "response.reasoning_summary_text.delta", "response.reasoning_text.delta": if event.Delta != "" { _, _ = a.reasoning.WriteString(event.Delta) @@ -550,15 +595,35 @@ func (a *BufferedResponseAccumulator) BuildOutput() []ResponsesOutput { return out } -// SupplementResponseOutput fills resp.Output from accumulated delta content -// when the terminal event delivered an empty output array. If resp.Output is -// already populated, this is a no-op (preserves backward compatibility). +// SupplementResponseOutput fills resp.Output from accumulated stream content +// when the terminal event delivered an empty output array. It also fills empty +// function-call arguments from authoritative argument-done events. func (a *BufferedResponseAccumulator) SupplementResponseOutput(resp *ResponsesResponse) { - if resp == nil || len(resp.Output) > 0 { + if resp == nil { return } - if !a.HasContent() { + if len(resp.Output) == 0 { + if a.HasContent() { + resp.Output = a.BuildOutput() + } return } - resp.Output = a.BuildOutput() + + for outputIndex := range resp.Output { + item := &resp.Output[outputIndex] + if item.Type != "function_call" || item.Arguments != "" { + continue + } + for funcIndex := range a.funcCalls { + call := &a.funcCalls[funcIndex] + matchesCallID := item.CallID != "" && item.CallID == call.CallID + if !matchesCallID && call.OutputIndex != outputIndex { + continue + } + if call.Args.Len() > 0 { + item.Arguments = call.Args.String() + } + break + } + } }